SERVER-50946 Create PlanExplainer interface and implementations

41 files changed, 1577 insertions, 1113 deletions

diff --git a/src/mongo/db/SConscript b/src/mongo/db/SConscript
index 2ad4e518b49..e813b2067c5 100644
--- a/src/mongo/db/SConscript
+++ b/src/mongo/db/SConscript
@@ -1170,6 +1170,7 @@ env.Library(
         'pipeline/document_source_geo_near_cursor.cpp',
         'pipeline/pipeline_d.cpp',
         'pipeline/plan_executor_pipeline.cpp',
+        'pipeline/plan_explainer_pipeline.cpp',
         'query/classic_stage_builder.cpp',
         'query/explain.cpp',
         'query/find.cpp',
@@ -1178,6 +1179,8 @@ env.Library(
         'query/plan_executor_impl.cpp',
         'query/plan_executor_factory.cpp',
         'query/plan_executor_sbe.cpp',
+        'query/plan_explainer_impl.cpp',
+        'query/plan_explainer_sbe.cpp',
         'query/plan_insert_listener.cpp',
         'query/plan_ranker.cpp',
         'query/plan_yield_policy_impl.cpp',
diff --git a/src/mongo/db/clientcursor.cpp b/src/mongo/db/clientcursor.cpp
index 33d41061774..56b4936cd6f 100644
--- a/src/mongo/db/clientcursor.cpp
+++ b/src/mongo/db/clientcursor.cpp
@@ -93,7 +93,7 @@ ClientCursor::ClientCursor(ClientCursorParams params,
       _operationUsingCursor(operationUsingCursor),
       _lastUseDate(now),
       _createdDate(now),
-      _planSummary(_exec->getPlanSummary()),
+      _planSummary(_exec->getPlanExplainer().getPlanSummary()),
       _opKey(operationUsingCursor->getOperationKey()) {
     invariant(_exec);
     invariant(_operationUsingCursor);
diff --git a/src/mongo/db/commands/count_cmd.cpp b/src/mongo/db/commands/count_cmd.cpp
index b15ecdc4383..427a9f2da96 100644
--- a/src/mongo/db/commands/count_cmd.cpp
+++ b/src/mongo/db/commands/count_cmd.cpp
@@ -264,20 +264,23 @@ public:
         auto curOp = CurOp::get(opCtx);
         {
             stdx::lock_guard<Client> lk(*opCtx->getClient());
-            curOp->setPlanSummary_inlock(exec->getPlanSummary());
+            curOp->setPlanSummary_inlock(exec->getPlanExplainer().getPlanSummary());
         }
 
         auto countResult = exec->executeCount();
 
         PlanSummaryStats summaryStats;
-        exec->getSummaryStats(&summaryStats);
+        exec->getPlanExplainer().getSummaryStats(&summaryStats);
         if (collection) {
             CollectionQueryInfo::get(collection).notifyOfQuery(opCtx, collection, summaryStats);
         }
         curOp->debug().setPlanSummaryMetrics(summaryStats);
 
         if (curOp->shouldDBProfile(opCtx)) {
-            curOp->debug().execStats = exec->getStats();
+            auto&& explainer = exec->getPlanExplainer();
+            auto&& [stats, _] =
+                explainer.getWinningPlanStats(ExplainOptions::Verbosity::kExecStats);
+            curOp->debug().execStats = std::move(stats);
         }
 
         result.appendNumber("n", countResult);
diff --git a/src/mongo/db/commands/distinct.cpp b/src/mongo/db/commands/distinct.cpp
index d6b76a3e7e0..226d3b52183 100644
--- a/src/mongo/db/commands/distinct.cpp
+++ b/src/mongo/db/commands/distinct.cpp
@@ -243,7 +243,8 @@ public:
 
         {
             stdx::lock_guard<Client> lk(*opCtx->getClient());
-            CurOp::get(opCtx)->setPlanSummary_inlock(executor.getValue()->getPlanSummary());
+            CurOp::get(opCtx)->setPlanSummary_inlock(
+                executor.getValue()->getPlanExplainer().getPlanSummary());
         }
 
         const auto key = cmdObj[ParsedDistinct::kKeyField].valuestrsafe();
@@ -284,12 +285,15 @@ public:
                 }
             }
         } catch (DBException& exception) {
+            auto&& explainer = executor.getValue()->getPlanExplainer();
+            auto&& [stats, _] =
+                explainer.getWinningPlanStats(ExplainOptions::Verbosity::kExecStats);
             LOGV2_WARNING(23797,
                           "Plan executor error during distinct command: {error}, "
                           "stats: {stats}",
                           "Plan executor error during distinct command",
                           "error"_attr = exception.toStatus(),
-                          "stats"_attr = redact(executor.getValue()->getStats()));
+                          "stats"_attr = redact(stats));
 
             exception.addContext("Executor error during distinct command");
             throw;
@@ -299,14 +303,17 @@ public:
 
         // Get summary information about the plan.
         PlanSummaryStats stats;
-        executor.getValue()->getSummaryStats(&stats);
+        auto&& explainer = executor.getValue()->getPlanExplainer();
+        explainer.getSummaryStats(&stats);
         if (collection) {
             CollectionQueryInfo::get(collection).notifyOfQuery(opCtx, collection, stats);
         }
         curOp->debug().setPlanSummaryMetrics(stats);
 
         if (curOp->shouldDBProfile(opCtx)) {
-            curOp->debug().execStats = executor.getValue()->getStats();
+            auto&& [stats, _] =
+                explainer.getWinningPlanStats(ExplainOptions::Verbosity::kExecStats);
+            curOp->debug().execStats = std::move(stats);
         }
 
         BSONArrayBuilder valueListBuilder(result.subarrayStart("values"));
diff --git a/src/mongo/db/commands/find_and_modify.cpp b/src/mongo/db/commands/find_and_modify.cpp
index 26851277138..4fe85f09807 100644
--- a/src/mongo/db/commands/find_and_modify.cpp
+++ b/src/mongo/db/commands/find_and_modify.cpp
@@ -96,11 +96,13 @@ boost::optional<BSONObj> advanceExecutor(OperationContext* opCtx,
     try {
         state = exec->getNext(&value, nullptr);
     } catch (DBException& exception) {
+        auto&& explainer = exec->getPlanExplainer();
+        auto&& [stats, _] = explainer.getWinningPlanStats(ExplainOptions::Verbosity::kExecStats);
         LOGV2_WARNING(23802,
                       "Plan executor error during findAndModify: {error}, stats: {stats}",
                       "Plan executor error during findAndModify",
                       "error"_attr = exception.toStatus(),
-                      "stats"_attr = redact(exec->getStats()));
+                      "stats"_attr = redact(stats));
 
         exception.addContext("Plan executor error during findAndModify");
         throw;
@@ -478,7 +480,7 @@ public:
 
         {
             stdx::lock_guard<Client> lk(*opCtx->getClient());
-            CurOp::get(opCtx)->setPlanSummary_inlock(exec->getPlanSummary());
+            CurOp::get(opCtx)->setPlanSummary_inlock(exec->getPlanExplainer().getPlanSummary());
         }
 
         auto docFound = advanceExecutor(opCtx, exec.get(), args.isRemove());
@@ -487,7 +489,7 @@ public:
         // multiple times.
 
         PlanSummaryStats summaryStats;
-        exec->getSummaryStats(&summaryStats);
+        exec->getPlanExplainer().getSummaryStats(&summaryStats);
         if (const auto& coll = collection.getCollection()) {
             CollectionQueryInfo::get(coll).notifyOfQuery(opCtx, coll, summaryStats);
         }
@@ -497,7 +499,10 @@ public:
         opDebug->additiveMetrics.ndeleted = docFound ? 1 : 0;
 
         if (curOp->shouldDBProfile(opCtx)) {
-            curOp->debug().execStats = exec->getStats();
+            auto&& explainer = exec->getPlanExplainer();
+            auto&& [stats, _] =
+                explainer.getWinningPlanStats(ExplainOptions::Verbosity::kExecStats);
+            curOp->debug().execStats = std::move(stats);
         }
         recordStatsForTopCommand(opCtx);
 
@@ -562,7 +567,7 @@ public:
 
         {
             stdx::lock_guard<Client> lk(*opCtx->getClient());
-            CurOp::get(opCtx)->setPlanSummary_inlock(exec->getPlanSummary());
+            CurOp::get(opCtx)->setPlanSummary_inlock(exec->getPlanExplainer().getPlanSummary());
         }
 
         auto docFound = advanceExecutor(opCtx, exec.get(), args.isRemove());
@@ -571,7 +576,8 @@ public:
         // multiple times.
 
         PlanSummaryStats summaryStats;
-        exec->getSummaryStats(&summaryStats);
+        auto&& explainer = exec->getPlanExplainer();
+        explainer.getSummaryStats(&summaryStats);
         if (collection) {
             CollectionQueryInfo::get(collection).notifyOfQuery(opCtx, collection, summaryStats);
         }
@@ -579,7 +585,9 @@ public:
         opDebug->setPlanSummaryMetrics(summaryStats);
 
         if (curOp->shouldDBProfile(opCtx)) {
-            curOp->debug().execStats = exec->getStats();
+            auto&& [stats, _] =
+                explainer.getWinningPlanStats(ExplainOptions::Verbosity::kExecStats);
+            curOp->debug().execStats = std::move(stats);
         }
         recordStatsForTopCommand(opCtx);
 
diff --git a/src/mongo/db/commands/find_cmd.cpp b/src/mongo/db/commands/find_cmd.cpp
index 1459df1fe91..bf992594df1 100644
--- a/src/mongo/db/commands/find_cmd.cpp
+++ b/src/mongo/db/commands/find_cmd.cpp
@@ -427,7 +427,7 @@ public:
 
             {
                 stdx::lock_guard<Client> lk(*opCtx->getClient());
-                CurOp::get(opCtx)->setPlanSummary_inlock(exec->getPlanSummary());
+                CurOp::get(opCtx)->setPlanSummary_inlock(exec->getPlanExplainer().getPlanSummary());
             }
 
             if (!collection) {
@@ -476,12 +476,15 @@ public:
             } catch (DBException& exception) {
                 firstBatch.abandon();
 
+                auto&& explainer = exec->getPlanExplainer();
+                auto&& [stats, _] =
+                    explainer.getWinningPlanStats(ExplainOptions::Verbosity::kExecStats);
                 LOGV2_WARNING(23798,
                               "Plan executor error during find command: {error}, "
                               "stats: {stats}",
                               "Plan executor error during find command",
                               "error"_attr = exception.toStatus(),
-                              "stats"_attr = redact(exec->getStats()));
+                              "stats"_attr = redact(stats));
 
                 exception.addContext("Executor error during find command");
                 throw;
diff --git a/src/mongo/db/commands/getmore_cmd.cpp b/src/mongo/db/commands/getmore_cmd.cpp
index bf4a69408a7..54be8b56b81 100644
--- a/src/mongo/db/commands/getmore_cmd.cpp
+++ b/src/mongo/db/commands/getmore_cmd.cpp
@@ -336,11 +336,14 @@ public:
             } catch (DBException& exception) {
                 nextBatch->abandon();
 
+                auto&& explainer = exec->getPlanExplainer();
+                auto&& [stats, _] =
+                    explainer.getWinningPlanStats(ExplainOptions::Verbosity::kExecStats);
                 LOGV2_WARNING(20478,
                               "getMore command executor error: {error}, stats: {stats}",
                               "getMore command executor error",
                               "error"_attr = exception.toStatus(),
-                              "stats"_attr = redact(exec->getStats()));
+                              "stats"_attr = redact(stats));
 
                 exception.addContext("Executor error during getMore");
                 throw;
@@ -515,7 +518,7 @@ public:
             exec->reattachToOperationContext(opCtx);
             exec->restoreState(nullptr);
 
-            auto planSummary = exec->getPlanSummary();
+            auto planSummary = exec->getPlanExplainer().getPlanSummary();
             {
                 stdx::lock_guard<Client> lk(*opCtx->getClient());
                 curOp->setPlanSummary_inlock(planSummary);
@@ -560,7 +563,7 @@ public:
             // obtain these values we need to take a diff of the pre-execution and post-execution
             // metrics, as they accumulate over the course of a cursor's lifetime.
             PlanSummaryStats preExecutionStats;
-            exec->getSummaryStats(&preExecutionStats);
+            exec->getPlanExplainer().getSummaryStats(&preExecutionStats);
 
             // Mark this as an AwaitData operation if appropriate.
             if (cursorPin->isAwaitData() && !disableAwaitDataFailpointActive) {
@@ -607,7 +610,7 @@ public:
                                                         &numResults);
 
             PlanSummaryStats postExecutionStats;
-            exec->getSummaryStats(&postExecutionStats);
+            exec->getPlanExplainer().getSummaryStats(&postExecutionStats);
             postExecutionStats.totalKeysExamined -= preExecutionStats.totalKeysExamined;
             postExecutionStats.totalDocsExamined -= preExecutionStats.totalDocsExamined;
             curOp->debug().setPlanSummaryMetrics(postExecutionStats);
@@ -620,7 +623,10 @@ public:
             if (cursorPin->getExecutor()->lockPolicy() !=
                     PlanExecutor::LockPolicy::kLocksInternally &&
                 curOp->shouldDBProfile(opCtx)) {
-                curOp->debug().execStats = exec->getStats();
+                auto&& explainer = exec->getPlanExplainer();
+                auto&& [stats, _] =
+                    explainer.getWinningPlanStats(ExplainOptions::Verbosity::kExecStats);
+                curOp->debug().execStats = std::move(stats);
             }
 
             if (shouldSaveCursor) {
diff --git a/src/mongo/db/commands/map_reduce_agg.cpp b/src/mongo/db/commands/map_reduce_agg.cpp
index 2f5b656262e..84ec6647e60 100644
--- a/src/mongo/db/commands/map_reduce_agg.cpp
+++ b/src/mongo/db/commands/map_reduce_agg.cpp
@@ -49,8 +49,10 @@
 #include "mongo/db/pipeline/document_source_cursor.h"
 #include "mongo/db/pipeline/expression.h"
 #include "mongo/db/pipeline/pipeline_d.h"
+#include "mongo/db/pipeline/plan_executor_pipeline.h"
 #include "mongo/db/query/explain_common.h"
 #include "mongo/db/query/map_reduce_output_format.h"
+#include "mongo/db/query/plan_executor_factory.h"
 
 namespace mongo::map_reduce_agg {
 
@@ -110,10 +112,12 @@ bool runAggregationMapReduce(OperationContext* opCtx,
                              const BSONObj& cmd,
                              BSONObjBuilder& result,
                              boost::optional<ExplainOptions::Verbosity> verbosity) {
-    auto exhaustPipelineIntoBSONArray = [](auto&& pipeline) {
+    auto exhaustPipelineIntoBSONArray = [](auto&& exec) {
         BSONArrayBuilder bab;
-        while (auto&& doc = pipeline->getNext())
-            bab.append(doc->toBson());
+        BSONObj obj;
+        while (exec->getNext(&obj, nullptr) == PlanExecutor::ADVANCED) {
+            bab.append(obj);
+        }
         return bab.arr();
     };
 
@@ -126,24 +130,25 @@ bool runAggregationMapReduce(OperationContext* opCtx,
         return expCtx->mongoProcessInterface->attachCursorSourceToPipelineForLocalRead(
             pipeline.release());
     }();
+    auto exec = plan_executor_factory::make(
+        expCtx, std::move(runnablePipeline), false /* isChangeStream */);
+    auto&& explainer = exec->getPlanExplainer();
 
     {
-        auto planSummaryStr = PipelineD::getPlanSummaryStr(runnablePipeline.get());
-
         stdx::lock_guard<Client> lk(*opCtx->getClient());
-        CurOp::get(opCtx)->setPlanSummary_inlock(std::move(planSummaryStr));
+        CurOp::get(opCtx)->setPlanSummary_inlock(explainer.getPlanSummary());
     }
 
     try {
-        auto resultArray = exhaustPipelineIntoBSONArray(runnablePipeline);
+        auto resultArray = exhaustPipelineIntoBSONArray(exec);
 
         if (expCtx->explain) {
-            result << "stages" << Value(runnablePipeline->writeExplainOps(*(expCtx->explain)));
-            explain_common::generateServerInfo(&result);
+            Explain::explainPipeline(
+                exec.get(), false /* executePipeline  */, *expCtx->explain, &result);
         }
 
         PlanSummaryStats planSummaryStats;
-        PipelineD::getPlanSummaryStats(runnablePipeline.get(), &planSummaryStats);
+        explainer.getSummaryStats(&planSummaryStats);
         CurOp::get(opCtx)->debug().setPlanSummaryMetrics(planSummaryStats);
 
         if (!expCtx->explain) {
diff --git a/src/mongo/db/commands/run_aggregate.cpp b/src/mongo/db/commands/run_aggregate.cpp
index 927b93d9e96..2bd30508ebd 100644
--- a/src/mongo/db/commands/run_aggregate.cpp
+++ b/src/mongo/db/commands/run_aggregate.cpp
@@ -182,11 +182,14 @@ bool handleCursorCommand(OperationContext* opCtx,
             exec = nullptr;
             break;
         } catch (DBException& exception) {
+            auto&& explainer = exec->getPlanExplainer();
+            auto&& [stats, _] =
+                explainer.getWinningPlanStats(ExplainOptions::Verbosity::kExecStats);
             LOGV2_WARNING(23799,
                           "Aggregate command executor error: {error}, stats: {stats}",
                           "Aggregate command executor error",
                           "error"_attr = exception.toStatus(),
-                          "stats"_attr = redact(exec->getStats()));
+                          "stats"_attr = redact(stats));
 
             exception.addContext("PlanExecutor error during aggregation");
             throw;
@@ -701,7 +704,7 @@ Status runAggregate(OperationContext* opCtx,
         }
 
         {
-            auto planSummary = execs[0]->getPlanSummary();
+            auto planSummary = execs[0]->getPlanExplainer().getPlanSummary();
             stdx::lock_guard<Client> lk(*opCtx->getClient());
             curOp->setPlanSummary_inlock(std::move(planSummary));
         }
@@ -752,7 +755,8 @@ Status runAggregate(OperationContext* opCtx,
         auto explainExecutor = pins[0]->getExecutor();
         auto bodyBuilder = result->getBodyBuilder();
         if (auto pipelineExec = dynamic_cast<PlanExecutorPipeline*>(explainExecutor)) {
-            Explain::explainPipelineExecutor(pipelineExec, *(expCtx->explain), &bodyBuilder);
+            Explain::explainPipeline(
+                pipelineExec, true /* executePipeline */, *(expCtx->explain), &bodyBuilder);
         } else {
             invariant(explainExecutor->getOpCtx() == opCtx);
             // The explainStages() function for a non-pipeline executor may need to execute the plan
@@ -775,7 +779,7 @@ Status runAggregate(OperationContext* opCtx,
         }
 
         PlanSummaryStats stats;
-        pins[0].getCursor()->getExecutor()->getSummaryStats(&stats);
+        pins[0].getCursor()->getExecutor()->getPlanExplainer().getSummaryStats(&stats);
         curOp->debug().setPlanSummaryMetrics(stats);
         curOp->debug().nreturned = stats.nReturned;
         // For an optimized away pipeline, signal the cache that a query operation has completed.
diff --git a/src/mongo/db/exec/cached_plan.cpp b/src/mongo/db/exec/cached_plan.cpp
index f0c6181d2de..842a2adf7f4 100644
--- a/src/mongo/db/exec/cached_plan.cpp
+++ b/src/mongo/db/exec/cached_plan.cpp
@@ -111,11 +111,12 @@ Status CachedPlanStage::pickBestPlan(PlanYieldPolicy* yieldPolicy) {
             // that a different plan is less resource-intensive, so we fall back to replanning the
             // whole query. We neither evict the existing cache entry nor cache the result of
             // replanning.
+            auto explainer = plan_explainer_factory::makePlanExplainer(child().get(), nullptr);
             LOGV2_DEBUG(20579,
                         1,
                         "Execution of cached plan failed, falling back to replan",
                         "query"_attr = redact(_canonicalQuery->toStringShort()),
-                        "planSummary"_attr = Explain::getPlanSummary(child().get()),
+                        "planSummary"_attr = explainer->getPlanSummary(),
                         "status"_attr = redact(ex.toStatus()));
 
             const bool shouldCache = false;
@@ -159,13 +160,14 @@ Status CachedPlanStage::pickBestPlan(PlanYieldPolicy* yieldPolicy) {
 
     // If we're here, the trial period took more than 'maxWorksBeforeReplan' work cycles. This
     // plan is taking too long, so we replan from scratch.
+    auto explainer = plan_explainer_factory::makePlanExplainer(child().get(), nullptr);
     LOGV2_DEBUG(20580,
                 1,
                 "Evicting cache entry and replanning query",
                 "maxWorksBeforeReplan"_attr = maxWorksBeforeReplan,
                 "decisionWorks"_attr = _decisionWorks,
                 "query"_attr = redact(_canonicalQuery->toStringShort()),
-                "planSummary"_attr = Explain::getPlanSummary(child().get()));
+                "planSummary"_attr = explainer->getPlanSummary());
 
     const bool shouldCache = true;
     return replan(
@@ -225,12 +227,13 @@ Status CachedPlanStage::replan(PlanYieldPolicy* yieldPolicy, bool shouldCache, s
         _replannedQs = std::move(solutions.back());
         solutions.pop_back();
 
+        auto explainer = plan_explainer_factory::makePlanExplainer(child().get(), nullptr);
         LOGV2_DEBUG(
             20581,
             1,
             "Replanning of query resulted in single query solution, which will not be cached.",
             "query"_attr = redact(_canonicalQuery->toStringShort()),
-            "planSummary"_attr = Explain::getPlanSummary(child().get()),
+            "planSummary"_attr = explainer->getPlanSummary(),
             "shouldCache"_attr = (shouldCache ? "yes" : "no"));
         return Status::OK();
     }
@@ -259,11 +262,12 @@ Status CachedPlanStage::replan(PlanYieldPolicy* yieldPolicy, bool shouldCache, s
         return pickBestPlanStatus;
     }
 
+    auto explainer = plan_explainer_factory::makePlanExplainer(child().get(), nullptr);
     LOGV2_DEBUG(20582,
                 1,
                 "Query plan after replanning and its cache status",
                 "query"_attr = redact(_canonicalQuery->toStringShort()),
-                "planSummary"_attr = Explain::getPlanSummary(child().get()),
+                "planSummary"_attr = explainer->getPlanSummary(),
                 "shouldCache"_attr = (shouldCache ? "yes" : "no"));
     return Status::OK();
 }
diff --git a/src/mongo/db/exec/multi_plan.cpp b/src/mongo/db/exec/multi_plan.cpp
index a9e985cd533..15e025ff907 100644
--- a/src/mongo/db/exec/multi_plan.cpp
+++ b/src/mongo/db/exec/multi_plan.cpp
@@ -199,8 +199,9 @@ Status MultiPlanStage::pickBestPlan(PlanYieldPolicy* yieldPolicy) {
 
     LOGV2_DEBUG(
         20590, 5, "Winning solution", "bestSolution"_attr = redact(bestSolution->toString()));
-    LOGV2_DEBUG(
-        20591, 2, "Winning plan", "planSummary"_attr = Explain::getPlanSummary(bestCandidate.root));
+
+    auto explainer = plan_explainer_factory::makePlanExplainer(bestCandidate.root, nullptr);
+    LOGV2_DEBUG(20591, 2, "Winning plan", "planSummary"_attr = explainer->getPlanSummary());
 
     _backupPlanIdx = kNoSuchPlan;
     if (bestSolution->hasBlockingStage && (0 == alreadyProduced.size())) {
diff --git a/src/mongo/db/exec/plan_cache_util.h b/src/mongo/db/exec/plan_cache_util.h
index 14946877777..ec5ca8a0358 100644
--- a/src/mongo/db/exec/plan_cache_util.h
+++ b/src/mongo/db/exec/plan_cache_util.h
@@ -110,21 +110,26 @@ void updatePlanCache(
             size_t winnerIdx = ranking->candidateOrder[0];
             size_t runnerUpIdx = ranking->candidateOrder[1];
 
+            auto winnerExplainer = plan_explainer_factory::makePlanExplainer(
+                &*candidates[winnerIdx].root, candidates[winnerIdx].solution.get());
+            auto runnerUpExplainer = plan_explainer_factory::makePlanExplainer(
+                &*candidates[runnerUpIdx].root, candidates[runnerUpIdx].solution.get());
+
             log_detail::logTieForBest(query.toStringShort(),
                                       ranking->scores[0],
                                       ranking->scores[1],
-                                      Explain::getPlanSummary(&*candidates[winnerIdx].root),
-                                      Explain::getPlanSummary(&*candidates[runnerUpIdx].root));
+                                      winnerExplainer->getPlanSummary(),
+                                      runnerUpExplainer->getPlanSummary());
         }
 
         if (candidates[winnerIdx].results.empty()) {
             // We're using the "sometimes cache" mode, and the winning plan produced no results
             // during the plan ranking trial period. We will not write a plan cache entry.
             canCache = false;
+            auto winnerExplainer = plan_explainer_factory::makePlanExplainer(
+                &*candidates[winnerIdx].root, candidates[winnerIdx].solution.get());
             log_detail::logNotCachingZeroResults(
-                query.toStringShort(),
-                ranking->scores[0],
-                Explain::getPlanSummary(&*candidates[winnerIdx].root));
+                query.toStringShort(), ranking->scores[0], winnerExplainer->getPlanSummary());
         }
     }
 
diff --git a/src/mongo/db/ops/write_ops_exec.cpp b/src/mongo/db/ops/write_ops_exec.cpp
index d0cdf8a29cd..981f1a542a4 100644
--- a/src/mongo/db/ops/write_ops_exec.cpp
+++ b/src/mongo/db/ops/write_ops_exec.cpp
@@ -670,19 +670,21 @@ static SingleWriteResult performSingleUpdateOp(OperationContext* opCtx,
 
     {
         stdx::lock_guard<Client> lk(*opCtx->getClient());
-        CurOp::get(opCtx)->setPlanSummary_inlock(exec->getPlanSummary());
+        CurOp::get(opCtx)->setPlanSummary_inlock(exec->getPlanExplainer().getPlanSummary());
     }
 
     auto updateResult = exec->executeUpdate();
 
     PlanSummaryStats summary;
-    exec->getSummaryStats(&summary);
+    auto&& explainer = exec->getPlanExplainer();
+    explainer.getSummaryStats(&summary);
     if (const auto& coll = collection->getCollection()) {
         CollectionQueryInfo::get(coll).notifyOfQuery(opCtx, coll, summary);
     }
 
     if (curOp.shouldDBProfile(opCtx)) {
-        curOp.debug().execStats = exec->getStats();
+        auto&& [stats, _] = explainer.getWinningPlanStats(ExplainOptions::Verbosity::kExecStats);
+        curOp.debug().execStats = std::move(stats);
     }
 
     recordUpdateResultInOpDebug(updateResult, &curOp.debug());
@@ -907,21 +909,23 @@ static SingleWriteResult performSingleDeleteOp(OperationContext* opCtx,
 
     {
         stdx::lock_guard<Client> lk(*opCtx->getClient());
-        CurOp::get(opCtx)->setPlanSummary_inlock(exec->getPlanSummary());
+        CurOp::get(opCtx)->setPlanSummary_inlock(exec->getPlanExplainer().getPlanSummary());
     }
 
     auto nDeleted = exec->executeDelete();
     curOp.debug().additiveMetrics.ndeleted = nDeleted;
 
     PlanSummaryStats summary;
-    exec->getSummaryStats(&summary);
+    auto&& explainer = exec->getPlanExplainer();
+    explainer.getSummaryStats(&summary);
     if (const auto& coll = collection.getCollection()) {
         CollectionQueryInfo::get(coll).notifyOfQuery(opCtx, coll, summary);
     }
     curOp.debug().setPlanSummaryMetrics(summary);
 
     if (curOp.shouldDBProfile(opCtx)) {
-        curOp.debug().execStats = exec->getStats();
+        auto&& [stats, _] = explainer.getWinningPlanStats(ExplainOptions::Verbosity::kExecStats);
+        curOp.debug().execStats = std::move(stats);
     }
 
     LastError::get(opCtx->getClient()).recordDelete(nDeleted);
diff --git a/src/mongo/db/pipeline/document_source_cursor.cpp b/src/mongo/db/pipeline/document_source_cursor.cpp
index a651b387ce4..c0ae3d1eb10 100644
--- a/src/mongo/db/pipeline/document_source_cursor.cpp
+++ b/src/mongo/db/pipeline/document_source_cursor.cpp
@@ -195,7 +195,7 @@ void DocumentSourceCursor::_updateOplogTimestamp() {
 
 void DocumentSourceCursor::recordPlanSummaryStats() {
     invariant(_exec);
-    _exec->getSummaryStats(&_planSummaryStats);
+    _exec->getPlanExplainer().getSummaryStats(&_planSummaryStats);
 }
 
 Value DocumentSourceCursor::serialize(boost::optional<ExplainOptions::Verbosity> verbosity) const {
@@ -226,7 +226,7 @@ Value DocumentSourceCursor::serialize(boost::optional<ExplainOptions::Verbosity>
                                collection,
                                verbosity.get(),
                                _execStatus,
-                               _winningPlanTrialStats.get(),
+                               _winningPlanTrialStats,
                                BSONObj(),
                                &explainStatsBuilder);
     }
@@ -300,13 +300,15 @@ DocumentSourceCursor::DocumentSourceCursor(
     // Later code in the DocumentSourceCursor lifecycle expects that '_exec' is in a saved state.
     _exec->saveState();
 
-    _planSummary = _exec->getPlanSummary();
+    auto&& explainer = _exec->getPlanExplainer();
+    _planSummary = explainer.getPlanSummary();
     recordPlanSummaryStats();
 
     if (pExpCtx->explain) {
         // It's safe to access the executor even if we don't have the collection lock since we're
         // just going to call getStats() on it.
-        _winningPlanTrialStats = Explain::getWinningPlanTrialStats(_exec.get());
+        _winningPlanTrialStats =
+            explainer.getWinningPlanStats(ExplainOptions::Verbosity::kExecStats);
     }
 
     if (collection) {
diff --git a/src/mongo/db/pipeline/document_source_cursor.h b/src/mongo/db/pipeline/document_source_cursor.h
index 4ee177593c5..360078f90a9 100644
--- a/src/mongo/db/pipeline/document_source_cursor.h
+++ b/src/mongo/db/pipeline/document_source_cursor.h
@@ -231,10 +231,10 @@ private:
     std::string _planSummary;
     PlanSummaryStats _planSummaryStats;
 
-    // Used only for explain() queries. Stores the stats of the winning plan when _exec's root
-    // stage is a MultiPlanStage. When the query is executed (with exec->executePlan()), it will
+    // Used only for explain() queries. Stores the stats of the winning plan when a plan was
+    // selected by the multi-planner. When the query is executed (with exec->executePlan()), it will
     // wipe out its own copy of the winning plan's statistics, so they need to be saved here.
-    std::unique_ptr<PlanStageStats> _winningPlanTrialStats;
+    boost::optional<PlanExplainer::PlanStatsDetails> _winningPlanTrialStats;
 
     // True if we are tracking the latest observed oplog timestamp, false otherwise.
     bool _trackOplogTS = false;
diff --git a/src/mongo/db/pipeline/pipeline_d.cpp b/src/mongo/db/pipeline/pipeline_d.cpp
index 6d0406f7495..e42bba7c620 100644
--- a/src/mongo/db/pipeline/pipeline_d.cpp
+++ b/src/mongo/db/pipeline/pipeline_d.cpp
@@ -761,32 +761,4 @@ Timestamp PipelineD::getLatestOplogTimestamp(const Pipeline* pipeline) {
     }
     return Timestamp();
 }
-
-std::string PipelineD::getPlanSummaryStr(const Pipeline* pipeline) {
-    if (auto docSourceCursor =
-            dynamic_cast<DocumentSourceCursor*>(pipeline->_sources.front().get())) {
-        return docSourceCursor->getPlanSummaryStr();
-    }
-
-    return "";
-}
-
-void PipelineD::getPlanSummaryStats(const Pipeline* pipeline, PlanSummaryStats* statsOut) {
-    invariant(statsOut);
-
-    if (auto docSourceCursor =
-            dynamic_cast<DocumentSourceCursor*>(pipeline->_sources.front().get())) {
-        *statsOut = docSourceCursor->getPlanSummaryStats();
-    }
-
-    for (auto&& source : pipeline->_sources) {
-        if (dynamic_cast<DocumentSourceSort*>(source.get()))
-            statsOut->hasSortStage = true;
-
-        statsOut->usedDisk = statsOut->usedDisk || source->usedDisk();
-        if (statsOut->usedDisk && statsOut->hasSortStage)
-            break;
-    }
-}
-
 }  // namespace mongo
diff --git a/src/mongo/db/pipeline/pipeline_d.h b/src/mongo/db/pipeline/pipeline_d.h
index 96a2b83c57c..a77945dd4a6 100644
--- a/src/mongo/db/pipeline/pipeline_d.h
+++ b/src/mongo/db/pipeline/pipeline_d.h
@@ -118,11 +118,6 @@ public:
                                                            const AggregationRequest* aggRequest,
                                                            Pipeline* pipeline);
 
-
-    static std::string getPlanSummaryStr(const Pipeline* pipeline);
-
-    static void getPlanSummaryStats(const Pipeline* pipeline, PlanSummaryStats* statsOut);
-
     static Timestamp getLatestOplogTimestamp(const Pipeline* pipeline);
 
     /**
diff --git a/src/mongo/db/pipeline/plan_executor_pipeline.cpp b/src/mongo/db/pipeline/plan_executor_pipeline.cpp
index 0556aacac06..3fcdc71cf61 100644
--- a/src/mongo/db/pipeline/plan_executor_pipeline.cpp
+++ b/src/mongo/db/pipeline/plan_executor_pipeline.cpp
@@ -32,6 +32,7 @@
 #include "mongo/db/pipeline/plan_executor_pipeline.h"
 
 #include "mongo/db/pipeline/pipeline_d.h"
+#include "mongo/db/pipeline/plan_explainer_pipeline.h"
 #include "mongo/db/pipeline/resume_token.h"
 #include "mongo/db/repl/speculative_majority_read_info.h"
 
@@ -40,7 +41,10 @@ namespace mongo {
 PlanExecutorPipeline::PlanExecutorPipeline(boost::intrusive_ptr<ExpressionContext> expCtx,
                                            std::unique_ptr<Pipeline, PipelineDeleter> pipeline,
                                            bool isChangeStream)
-    : _expCtx(std::move(expCtx)), _pipeline(std::move(pipeline)), _isChangeStream(isChangeStream) {
+    : _expCtx(std::move(expCtx)),
+      _pipeline(std::move(pipeline)),
+      _planExplainer{_pipeline.get()},
+      _isChangeStream(isChangeStream) {
     // Pipeline plan executors must always have an ExpressionContext.
     invariant(_expCtx);
 
@@ -69,7 +73,7 @@ PlanExecutor::ExecState PlanExecutorPipeline::getNext(BSONObj* objOut, RecordId*
     if (!_stash.empty()) {
         *objOut = std::move(_stash.front());
         _stash.pop();
-        ++_nReturned;
+        _planExplainer.incrementNReturned();
         return PlanExecutor::ADVANCED;
     }
 
@@ -96,7 +100,7 @@ PlanExecutor::ExecState PlanExecutorPipeline::getNextDocument(Document* docOut,
 
     if (auto next = _getNext()) {
         *docOut = std::move(*next);
-        ++_nReturned;
+        _planExplainer.incrementNReturned();
         return PlanExecutor::ADVANCED;
     }
 
@@ -148,16 +152,6 @@ void PlanExecutorPipeline::_performChangeStreamsAccounting(const boost::optional
     }
 }
 
-std::string PlanExecutorPipeline::getPlanSummary() const {
-    return PipelineD::getPlanSummaryStr(_pipeline.get());
-}
-
-void PlanExecutorPipeline::getSummaryStats(PlanSummaryStats* statsOut) const {
-    invariant(statsOut);
-    PipelineD::getPlanSummaryStats(_pipeline.get(), statsOut);
-    statsOut->nReturned = _nReturned;
-}
-
 void PlanExecutorPipeline::_validateResumeToken(const Document& event) const {
     // If we are producing output to be merged on mongoS, then no stages can have modified the _id.
     if (_expCtx->needsMerge) {
diff --git a/src/mongo/db/pipeline/plan_executor_pipeline.h b/src/mongo/db/pipeline/plan_executor_pipeline.h
index 72d61471720..2bd7ae78041 100644
--- a/src/mongo/db/pipeline/plan_executor_pipeline.h
+++ b/src/mongo/db/pipeline/plan_executor_pipeline.h
@@ -33,6 +33,7 @@
 
 #include "mongo/db/exec/document_value/document.h"
 #include "mongo/db/pipeline/pipeline.h"
+#include "mongo/db/pipeline/plan_explainer_pipeline.h"
 #include "mongo/db/query/plan_executor.h"
 
 namespace mongo {
@@ -127,9 +128,9 @@ public:
         return LockPolicy::kLocksInternally;
     }
 
-    std::string getPlanSummary() const override;
-
-    void getSummaryStats(PlanSummaryStats* statsOut) const override;
+    const PlanExplainer& getPlanExplainer() const final {
+        return _planExplainer;
+    }
 
     /**
      * Writes the explain information about the underlying pipeline to a std::vector<Value>,
@@ -139,11 +140,6 @@ public:
         return _pipeline->writeExplainOps(verbosity);
     }
 
-    BSONObj getStats() const override {
-        // TODO SERVER-49808: Report execution stats for the pipeline.
-        return BSONObj{};
-    }
-
 private:
     /**
      * Obtains the next document from the underlying Pipeline, and does change streams-related
@@ -172,6 +168,8 @@ private:
 
     std::unique_ptr<Pipeline, PipelineDeleter> _pipeline;
 
+    PlanExplainerPipeline _planExplainer;
+
     const bool _isChangeStream;
 
     std::queue<BSONObj> _stash;
@@ -180,8 +178,6 @@ private:
     // reason for the kill.
     Status _killStatus = Status::OK();
 
-    size_t _nReturned = 0;
-
     // Set to true once we have received all results from the underlying '_pipeline', and the
     // pipeline has indicated end-of-stream.
     bool _pipelineIsEof = false;
diff --git a/src/mongo/db/pipeline/plan_explainer_pipeline.cpp b/src/mongo/db/pipeline/plan_explainer_pipeline.cpp
new file mode 100644
index 00000000000..b5ecf484144
--- /dev/null
+++ b/src/mongo/db/pipeline/plan_explainer_pipeline.cpp
@@ -0,0 +1,88 @@
+/**
+ *    Copyright (C) 2020-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/db/pipeline/plan_explainer_pipeline.h"
+
+#include "mongo/db/pipeline/document_source_cursor.h"
+#include "mongo/db/pipeline/document_source_sort.h"
+#include "mongo/db/pipeline/plan_executor_pipeline.h"
+#include "mongo/db/query/explain.h"
+
+namespace mongo {
+std::string PlanExplainerPipeline::getPlanSummary() const {
+    if (auto docSourceCursor =
+            dynamic_cast<DocumentSourceCursor*>(_pipeline->getSources().front().get())) {
+        return docSourceCursor->getPlanSummaryStr();
+    }
+
+    return "";
+}
+
+void PlanExplainerPipeline::getSummaryStats(PlanSummaryStats* statsOut) const {
+    invariant(statsOut);
+
+    if (auto docSourceCursor =
+            dynamic_cast<DocumentSourceCursor*>(_pipeline->getSources().front().get())) {
+        *statsOut = docSourceCursor->getPlanSummaryStats();
+    }
+
+    for (auto&& source : _pipeline->getSources()) {
+        if (dynamic_cast<DocumentSourceSort*>(source.get()))
+            statsOut->hasSortStage = true;
+
+        statsOut->usedDisk = statsOut->usedDisk || source->usedDisk();
+        if (statsOut->usedDisk && statsOut->hasSortStage)
+            break;
+    }
+
+    if (_nReturned) {
+        statsOut->nReturned = _nReturned;
+    }
+}
+
+PlanExplainer::PlanStatsDetails PlanExplainerPipeline::getWinningPlanStats(
+    ExplainOptions::Verbosity verbosity) const {
+    // TODO SERVER-49808: Report execution stats for the pipeline.
+    return {};
+}
+
+std::vector<PlanExplainer::PlanStatsDetails> PlanExplainerPipeline::getRejectedPlansStats(
+    ExplainOptions::Verbosity verbosity) const {
+    // Multi-planning is not supported for aggregation pipelines.
+    return {};
+}
+
+std::vector<PlanExplainer::PlanStatsDetails> PlanExplainerPipeline::getCachedPlanStats(
+    const PlanCacheEntry& entry, ExplainOptions::Verbosity verbosity) const {
+    // Pipelines are not cached, so we should never try to rebuild the stats from a cached entry.
+    MONGO_UNREACHABLE;
+}
+}  // namespace mongo
diff --git a/src/mongo/db/pipeline/plan_explainer_pipeline.h b/src/mongo/db/pipeline/plan_explainer_pipeline.h
new file mode 100644
index 00000000000..bf31b2c33a5
--- /dev/null
+++ b/src/mongo/db/pipeline/plan_explainer_pipeline.h
@@ -0,0 +1,63 @@
+/**
+ *    Copyright (C) 2020-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include "mongo/db/pipeline/pipeline.h"
+#include "mongo/db/query/plan_explainer.h"
+
+namespace mongo {
+/**
+ * A PlanExplainer implementation for aggregation pipelines.
+ */
+class PlanExplainerPipeline final : public PlanExplainer {
+public:
+    PlanExplainerPipeline(const Pipeline* pipeline) : _pipeline{pipeline} {}
+
+    bool isMultiPlan() const final {
+        return false;
+    }
+
+    std::string getPlanSummary() const final;
+    void getSummaryStats(PlanSummaryStats* statsOut) const final;
+    PlanStatsDetails getWinningPlanStats(ExplainOptions::Verbosity verbosity) const final;
+    std::vector<PlanStatsDetails> getRejectedPlansStats(
+        ExplainOptions::Verbosity verbosity) const final;
+    std::vector<PlanStatsDetails> getCachedPlanStats(
+        const PlanCacheEntry& entry, ExplainOptions::Verbosity verbosity) const final;
+
+    void incrementNReturned() {
+        ++_nReturned;
+    }
+
+private:
+    const Pipeline* const _pipeline;
+    size_t _nReturned{0};
+};
+}  // namespace mongo
diff --git a/src/mongo/db/query/explain.cpp b/src/mongo/db/query/explain.cpp
index 003c5fe202c..af107700c0d 100644
--- a/src/mongo/db/query/explain.cpp
+++ b/src/mongo/db/query/explain.cpp
@@ -63,551 +63,30 @@
 #include "mongo/util/str.h"
 #include "mongo/util/version.h"
 
+namespace mongo {
 namespace {
-
-using namespace mongo;
-using std::string;
-using std::unique_ptr;
-using std::vector;
-
-/**
- * Traverse the tree rooted at 'root', and add all tree nodes into the list 'flattened'.
- */
-void flattenStatsTree(const PlanStageStats* root, vector<const PlanStageStats*>* flattened) {
-    invariant(root->stageType != STAGE_MULTI_PLAN);
-    flattened->push_back(root);
-    for (auto&& child : root->children) {
-        flattenStatsTree(child.get(), flattened);
-    }
-}
-
-/**
- * Adds to the plan summary string being built by 'sb' for the execution stage 'stage'.
- */
-void addStageSummaryStr(const PlanStage* stage, StringBuilder& sb) {
-    // First add the stage type string.
-    const CommonStats* common = stage->getCommonStats();
-    sb << common->stageTypeStr;
-
-    // Some leaf nodes also provide info about the index they used.
-    const SpecificStats* specific = stage->getSpecificStats();
-    if (STAGE_COUNT_SCAN == stage->stageType()) {
-        const CountScanStats* spec = static_cast<const CountScanStats*>(specific);
-        const KeyPattern keyPattern{spec->keyPattern};
-        sb << " " << keyPattern;
-    } else if (STAGE_DISTINCT_SCAN == stage->stageType()) {
-        const DistinctScanStats* spec = static_cast<const DistinctScanStats*>(specific);
-        const KeyPattern keyPattern{spec->keyPattern};
-        sb << " " << keyPattern;
-    } else if (STAGE_GEO_NEAR_2D == stage->stageType()) {
-        const NearStats* spec = static_cast<const NearStats*>(specific);
-        const KeyPattern keyPattern{spec->keyPattern};
-        sb << " " << keyPattern;
-    } else if (STAGE_GEO_NEAR_2DSPHERE == stage->stageType()) {
-        const NearStats* spec = static_cast<const NearStats*>(specific);
-        const KeyPattern keyPattern{spec->keyPattern};
-        sb << " " << keyPattern;
-    } else if (STAGE_IXSCAN == stage->stageType()) {
-        const IndexScanStats* spec = static_cast<const IndexScanStats*>(specific);
-        const KeyPattern keyPattern{spec->keyPattern};
-        sb << " " << keyPattern;
-    } else if (STAGE_TEXT == stage->stageType()) {
-        const TextStats* spec = static_cast<const TextStats*>(specific);
-        const KeyPattern keyPattern{spec->indexPrefix};
-        sb << " " << keyPattern;
-    }
-}
-
 /**
- * Adds the path-level multikey information to the explain output in a field called "multiKeyPaths".
- * The value associated with the "multiKeyPaths" field is an object with keys equal to those in the
- * index key pattern and values equal to an array of strings corresponding to paths that cause the
- * index to be multikey.
+ * Adds the 'queryPlanner' explain section to the BSON object being built by 'out'.
  *
- * For example, with the index {'a.b': 1, 'a.c': 1} where the paths "a" and "a.b" cause the
- * index to be multikey, we'd have {'multiKeyPaths': {'a.b': ['a', 'a.b'], 'a.c': ['a']}}.
+ * This is a helper for generating explain BSON. It is used by explainStages(...).
  *
- * This function should only be called if the associated index supports path-level multikey
- * tracking.
- */
-void appendMultikeyPaths(const BSONObj& keyPattern,
-                         const MultikeyPaths& multikeyPaths,
-                         BSONObjBuilder* bob) {
-    BSONObjBuilder subMultikeyPaths(bob->subobjStart("multiKeyPaths"));
-
-    size_t i = 0;
-    for (const auto& keyElem : keyPattern) {
-        const FieldRef path{keyElem.fieldNameStringData()};
-
-        BSONArrayBuilder arrMultikeyComponents(
-            subMultikeyPaths.subarrayStart(keyElem.fieldNameStringData()));
-        for (const auto& multikeyComponent : multikeyPaths[i]) {
-            arrMultikeyComponents.append(path.dottedSubstring(0, multikeyComponent + 1));
-        }
-        arrMultikeyComponents.doneFast();
-
-        ++i;
-    }
-
-    subMultikeyPaths.doneFast();
-}
-
-/**
- * Gather the PlanStageStats for all of the losing plans. If exec doesn't have a MultiPlanStage
- * (or any losing plans), will return an empty vector.
+ * - 'exec' is a PlanExecutor which executes the plan for the operation being explained.
+ * - 'collection' is the collection used in the operation. The caller should hold an IS lock on the
+ *    collection which the query is for, even if 'collection' is nullptr.
+ * - 'extraInfo' specifies additional information to include into the output.
+ * - 'out' is a builder for the explain output.
  */
-std::vector<std::unique_ptr<PlanStageStats>> getRejectedPlansTrialStats(PlanExecutorImpl* exec) {
-    // Inspect the tree to see if there is a MultiPlanStage. Plan selection has already happened at
-    // this point, since we have a PlanExecutor.
-    const auto mps = exec->getMultiPlanStage();
-    std::vector<std::unique_ptr<PlanStageStats>> res;
-
-    // Get the stats from the trial period for all the plans.
-    if (mps) {
-        const auto mpsStats = mps->getStats();
-        for (size_t i = 0; i < mpsStats->children.size(); ++i) {
-            if (i != static_cast<size_t>(mps->bestPlanIdx())) {
-                res.emplace_back(std::move(mpsStats->children[i]));
-            }
-        }
-    }
-
-    return res;
-}
-
-/**
- * Get PlanExecutor's winning plan stats tree.
- */
-unique_ptr<PlanStageStats> getWinningPlanStatsTree(const PlanExecutorImpl* exec) {
-    auto mps = exec->getMultiPlanStage();
-    return mps ? std::move(mps->getStats()->children[mps->bestPlanIdx()])
-               : std::move(exec->getRootStage()->getStats());
-}
-
-/**
- * Executes the given plan executor, discarding the resulting documents, until it reaches EOF. If a
- * runtime error occur or execution is killed, throws a DBException.
- *
- * If 'exec' is configured for yielding, then a call to this helper could result in a yield.
- */
-void executePlan(PlanExecutor* exec) {
-    BSONObj obj;
-    while (exec->getNext(&obj, nullptr) == PlanExecutor::ADVANCED) {
-        // Discard the resulting documents.
-    }
-}
-
-}  // namespace
-
-namespace mongo {
-
-using str::stream;
-
-void Explain::flattenExecTree(const PlanStage* root, vector<const PlanStage*>* flattened) {
-    flattened->push_back(root);
-
-    if (root->stageType() == STAGE_MULTI_PLAN) {
-        // Only add the winning plan from a MultiPlanStage.
-        auto mps = static_cast<const MultiPlanStage*>(root);
-        const PlanStage* winningStage = mps->getChildren()[mps->bestPlanIdx()].get();
-        return flattenExecTree(winningStage, flattened);
-    }
-
-    const auto& children = root->getChildren();
-    for (size_t i = 0; i < children.size(); ++i) {
-        flattenExecTree(children[i].get(), flattened);
-    }
-}
-
-size_t Explain::getKeysExamined(StageType type, const SpecificStats* specific) {
-    if (STAGE_IXSCAN == type) {
-        const IndexScanStats* spec = static_cast<const IndexScanStats*>(specific);
-        return spec->keysExamined;
-    } else if (STAGE_IDHACK == type) {
-        const IDHackStats* spec = static_cast<const IDHackStats*>(specific);
-        return spec->keysExamined;
-    } else if (STAGE_COUNT_SCAN == type) {
-        const CountScanStats* spec = static_cast<const CountScanStats*>(specific);
-        return spec->keysExamined;
-    } else if (STAGE_DISTINCT_SCAN == type) {
-        const DistinctScanStats* spec = static_cast<const DistinctScanStats*>(specific);
-        return spec->keysExamined;
-    }
-
-    return 0;
-}
-
-size_t Explain::getDocsExamined(StageType type, const SpecificStats* specific) {
-    if (STAGE_COLLSCAN == type) {
-        const CollectionScanStats* spec = static_cast<const CollectionScanStats*>(specific);
-        return spec->docsTested;
-    } else if (STAGE_FETCH == type) {
-        const FetchStats* spec = static_cast<const FetchStats*>(specific);
-        return spec->docsExamined;
-    } else if (STAGE_IDHACK == type) {
-        const IDHackStats* spec = static_cast<const IDHackStats*>(specific);
-        return spec->docsExamined;
-    } else if (STAGE_TEXT_OR == type) {
-        const TextOrStats* spec = static_cast<const TextOrStats*>(specific);
-        return spec->fetches;
-    }
-
-    return 0;
-}
-
-void Explain::statsToBSON(const PlanStageStats& stats,
-                          ExplainOptions::Verbosity verbosity,
-                          BSONObjBuilder* bob,
-                          BSONObjBuilder* topLevelBob) {
-    invariant(bob);
-    invariant(topLevelBob);
-
-    // Stop as soon as the BSON object we're building exceeds 10 MB.
-    static const int kMaxStatsBSONSize = 10 * 1024 * 1024;
-    if (topLevelBob->len() > kMaxStatsBSONSize) {
-        bob->append("warning", "stats tree exceeded 10 MB");
-        return;
-    }
-
-    // Stage name.
-    bob->append("stage", stats.common.stageTypeStr);
-
-    // Display the BSON representation of the filter, if there is one.
-    if (!stats.common.filter.isEmpty()) {
-        bob->append("filter", stats.common.filter);
-    }
-
-    // Some top-level exec stats get pulled out of the root stage.
-    if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
-        bob->appendNumber("nReturned", stats.common.advanced);
-        // Include executionTimeMillis if it was recorded.
-        if (stats.common.executionTimeMillis) {
-            bob->appendNumber("executionTimeMillisEstimate", *stats.common.executionTimeMillis);
-        }
-
-        bob->appendNumber("works", stats.common.works);
-        bob->appendNumber("advanced", stats.common.advanced);
-        bob->appendNumber("needTime", stats.common.needTime);
-        bob->appendNumber("needYield", stats.common.needYield);
-        bob->appendNumber("saveState", stats.common.yields);
-        bob->appendNumber("restoreState", stats.common.unyields);
-        if (stats.common.failed)
-            bob->appendBool("failed", stats.common.failed);
-        bob->appendNumber("isEOF", stats.common.isEOF);
-    }
-
-    // Stage-specific stats
-    if (STAGE_AND_HASH == stats.stageType) {
-        AndHashStats* spec = static_cast<AndHashStats*>(stats.specific.get());
-
-        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
-            bob->appendNumber("memUsage", spec->memUsage);
-            bob->appendNumber("memLimit", spec->memLimit);
-
-            for (size_t i = 0; i < spec->mapAfterChild.size(); ++i) {
-                bob->appendNumber(string(stream() << "mapAfterChild_" << i),
-                                  spec->mapAfterChild[i]);
-            }
-        }
-    } else if (STAGE_AND_SORTED == stats.stageType) {
-        AndSortedStats* spec = static_cast<AndSortedStats*>(stats.specific.get());
-
-        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
-            for (size_t i = 0; i < spec->failedAnd.size(); ++i) {
-                bob->appendNumber(string(stream() << "failedAnd_" << i), spec->failedAnd[i]);
-            }
-        }
-    } else if (STAGE_COLLSCAN == stats.stageType) {
-        CollectionScanStats* spec = static_cast<CollectionScanStats*>(stats.specific.get());
-        bob->append("direction", spec->direction > 0 ? "forward" : "backward");
-        if (spec->minTs) {
-            bob->append("minTs", *(spec->minTs));
-        }
-        if (spec->maxTs) {
-            bob->append("maxTs", *(spec->maxTs));
-        }
-        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
-            bob->appendNumber("docsExamined", spec->docsTested);
-        }
-    } else if (STAGE_COUNT == stats.stageType) {
-        CountStats* spec = static_cast<CountStats*>(stats.specific.get());
-
-        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
-            bob->appendNumber("nCounted", spec->nCounted);
-            bob->appendNumber("nSkipped", spec->nSkipped);
-        }
-    } else if (STAGE_COUNT_SCAN == stats.stageType) {
-        CountScanStats* spec = static_cast<CountScanStats*>(stats.specific.get());
-
-        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
-            bob->appendNumber("keysExamined", spec->keysExamined);
-        }
-
-        bob->append("keyPattern", spec->keyPattern);
-        bob->append("indexName", spec->indexName);
-        if (!spec->collation.isEmpty()) {
-            bob->append("collation", spec->collation);
-        }
-        bob->appendBool("isMultiKey", spec->isMultiKey);
-        if (!spec->multiKeyPaths.empty()) {
-            appendMultikeyPaths(spec->keyPattern, spec->multiKeyPaths, bob);
-        }
-        bob->appendBool("isUnique", spec->isUnique);
-        bob->appendBool("isSparse", spec->isSparse);
-        bob->appendBool("isPartial", spec->isPartial);
-        bob->append("indexVersion", spec->indexVersion);
-
-        BSONObjBuilder indexBoundsBob;
-        indexBoundsBob.append("startKey", spec->startKey);
-        indexBoundsBob.append("startKeyInclusive", spec->startKeyInclusive);
-        indexBoundsBob.append("endKey", spec->endKey);
-        indexBoundsBob.append("endKeyInclusive", spec->endKeyInclusive);
-        bob->append("indexBounds", indexBoundsBob.obj());
-    } else if (STAGE_DELETE == stats.stageType) {
-        DeleteStats* spec = static_cast<DeleteStats*>(stats.specific.get());
-
-        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
-            bob->appendNumber("nWouldDelete", spec->docsDeleted);
-        }
-    } else if (STAGE_DISTINCT_SCAN == stats.stageType) {
-        DistinctScanStats* spec = static_cast<DistinctScanStats*>(stats.specific.get());
-
-        bob->append("keyPattern", spec->keyPattern);
-        bob->append("indexName", spec->indexName);
-        if (!spec->collation.isEmpty()) {
-            bob->append("collation", spec->collation);
-        }
-        bob->appendBool("isMultiKey", spec->isMultiKey);
-        if (!spec->multiKeyPaths.empty()) {
-            appendMultikeyPaths(spec->keyPattern, spec->multiKeyPaths, bob);
-        }
-        bob->appendBool("isUnique", spec->isUnique);
-        bob->appendBool("isSparse", spec->isSparse);
-        bob->appendBool("isPartial", spec->isPartial);
-        bob->append("indexVersion", spec->indexVersion);
-        bob->append("direction", spec->direction > 0 ? "forward" : "backward");
-
-        if ((topLevelBob->len() + spec->indexBounds.objsize()) > kMaxStatsBSONSize) {
-            bob->append("warning", "index bounds omitted due to BSON size limit");
-        } else {
-            bob->append("indexBounds", spec->indexBounds);
-        }
-
-        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
-            bob->appendNumber("keysExamined", spec->keysExamined);
-        }
-    } else if (STAGE_ENSURE_SORTED == stats.stageType) {
-        EnsureSortedStats* spec = static_cast<EnsureSortedStats*>(stats.specific.get());
-
-        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
-            bob->appendNumber("nDropped", spec->nDropped);
-        }
-    } else if (STAGE_FETCH == stats.stageType) {
-        FetchStats* spec = static_cast<FetchStats*>(stats.specific.get());
-        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
-            bob->appendNumber("docsExamined", spec->docsExamined);
-            bob->appendNumber("alreadyHasObj", spec->alreadyHasObj);
-        }
-    } else if (STAGE_GEO_NEAR_2D == stats.stageType || STAGE_GEO_NEAR_2DSPHERE == stats.stageType) {
-        NearStats* spec = static_cast<NearStats*>(stats.specific.get());
-
-        bob->append("keyPattern", spec->keyPattern);
-        bob->append("indexName", spec->indexName);
-        bob->append("indexVersion", spec->indexVersion);
-
-        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
-            BSONArrayBuilder intervalsBob(bob->subarrayStart("searchIntervals"));
-            for (vector<IntervalStats>::const_iterator it = spec->intervalStats.begin();
-                 it != spec->intervalStats.end();
-                 ++it) {
-                BSONObjBuilder intervalBob(intervalsBob.subobjStart());
-                intervalBob.append("minDistance", it->minDistanceAllowed);
-                intervalBob.append("maxDistance", it->maxDistanceAllowed);
-                intervalBob.append("maxInclusive", it->inclusiveMaxDistanceAllowed);
-                intervalBob.appendNumber("nBuffered", it->numResultsBuffered);
-                intervalBob.appendNumber("nReturned", it->numResultsReturned);
-            }
-            intervalsBob.doneFast();
-        }
-    } else if (STAGE_IDHACK == stats.stageType) {
-        IDHackStats* spec = static_cast<IDHackStats*>(stats.specific.get());
-        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
-            bob->appendNumber("keysExamined", spec->keysExamined);
-            bob->appendNumber("docsExamined", spec->docsExamined);
-        }
-    } else if (STAGE_IXSCAN == stats.stageType) {
-        IndexScanStats* spec = static_cast<IndexScanStats*>(stats.specific.get());
-
-        bob->append("keyPattern", spec->keyPattern);
-        bob->append("indexName", spec->indexName);
-        if (!spec->collation.isEmpty()) {
-            bob->append("collation", spec->collation);
-        }
-        bob->appendBool("isMultiKey", spec->isMultiKey);
-        if (!spec->multiKeyPaths.empty()) {
-            appendMultikeyPaths(spec->keyPattern, spec->multiKeyPaths, bob);
-        }
-        bob->appendBool("isUnique", spec->isUnique);
-        bob->appendBool("isSparse", spec->isSparse);
-        bob->appendBool("isPartial", spec->isPartial);
-        bob->append("indexVersion", spec->indexVersion);
-        bob->append("direction", spec->direction > 0 ? "forward" : "backward");
-
-        if ((topLevelBob->len() + spec->indexBounds.objsize()) > kMaxStatsBSONSize) {
-            bob->append("warning", "index bounds omitted due to BSON size limit");
-        } else {
-            bob->append("indexBounds", spec->indexBounds);
-        }
-
-        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
-            bob->appendNumber("keysExamined", spec->keysExamined);
-            bob->appendNumber("seeks", spec->seeks);
-            bob->appendNumber("dupsTested", spec->dupsTested);
-            bob->appendNumber("dupsDropped", spec->dupsDropped);
-        }
-    } else if (STAGE_OR == stats.stageType) {
-        OrStats* spec = static_cast<OrStats*>(stats.specific.get());
-
-        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
-            bob->appendNumber("dupsTested", spec->dupsTested);
-            bob->appendNumber("dupsDropped", spec->dupsDropped);
-        }
-    } else if (STAGE_LIMIT == stats.stageType) {
-        LimitStats* spec = static_cast<LimitStats*>(stats.specific.get());
-        bob->appendNumber("limitAmount", spec->limit);
-    } else if (isProjectionStageType(stats.stageType)) {
-        ProjectionStats* spec = static_cast<ProjectionStats*>(stats.specific.get());
-        bob->append("transformBy", spec->projObj);
-    } else if (STAGE_RECORD_STORE_FAST_COUNT == stats.stageType) {
-        CountStats* spec = static_cast<CountStats*>(stats.specific.get());
-
-        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
-            bob->appendNumber("nCounted", spec->nCounted);
-            bob->appendNumber("nSkipped", spec->nSkipped);
-        }
-    } else if (STAGE_SHARDING_FILTER == stats.stageType) {
-        ShardingFilterStats* spec = static_cast<ShardingFilterStats*>(stats.specific.get());
-
-        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
-            bob->appendNumber("chunkSkips", spec->chunkSkips);
-        }
-    } else if (STAGE_SKIP == stats.stageType) {
-        SkipStats* spec = static_cast<SkipStats*>(stats.specific.get());
-        bob->appendNumber("skipAmount", spec->skip);
-    } else if (isSortStageType(stats.stageType)) {
-        SortStats* spec = static_cast<SortStats*>(stats.specific.get());
-        bob->append("sortPattern", spec->sortPattern);
-        bob->appendIntOrLL("memLimit", spec->maxMemoryUsageBytes);
-
-        if (spec->limit > 0) {
-            bob->appendIntOrLL("limitAmount", spec->limit);
-        }
-
-        bob->append("type", stats.stageType == STAGE_SORT_SIMPLE ? "simple" : "default");
-
-        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
-            bob->appendIntOrLL("totalDataSizeSorted", spec->totalDataSizeBytes);
-            bob->appendBool("usedDisk", spec->wasDiskUsed);
-        }
-    } else if (STAGE_SORT_MERGE == stats.stageType) {
-        MergeSortStats* spec = static_cast<MergeSortStats*>(stats.specific.get());
-        bob->append("sortPattern", spec->sortPattern);
-
-        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
-            bob->appendNumber("dupsTested", spec->dupsTested);
-            bob->appendNumber("dupsDropped", spec->dupsDropped);
-        }
-    } else if (STAGE_TEXT == stats.stageType) {
-        TextStats* spec = static_cast<TextStats*>(stats.specific.get());
-
-        bob->append("indexPrefix", spec->indexPrefix);
-        bob->append("indexName", spec->indexName);
-        bob->append("parsedTextQuery", spec->parsedTextQuery);
-        bob->append("textIndexVersion", spec->textIndexVersion);
-    } else if (STAGE_TEXT_MATCH == stats.stageType) {
-        TextMatchStats* spec = static_cast<TextMatchStats*>(stats.specific.get());
-
-        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
-            bob->appendNumber("docsRejected", spec->docsRejected);
-        }
-    } else if (STAGE_TEXT_OR == stats.stageType) {
-        TextOrStats* spec = static_cast<TextOrStats*>(stats.specific.get());
-
-        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
-            bob->appendNumber("docsExamined", spec->fetches);
-        }
-    } else if (STAGE_UPDATE == stats.stageType) {
-        UpdateStats* spec = static_cast<UpdateStats*>(stats.specific.get());
-
-        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
-            bob->appendNumber("nMatched", spec->nMatched);
-            bob->appendNumber("nWouldModify", spec->nModified);
-            bob->appendNumber("nWouldUpsert", spec->nUpserted);
-        }
-    }
-
-    // We're done if there are no children.
-    if (stats.children.empty()) {
-        return;
-    }
-
-    // If there's just one child (a common scenario), avoid making an array. This makes
-    // the output more readable by saving a level of nesting. Name the field 'inputStage'
-    // rather than 'inputStages'.
-    if (1 == stats.children.size()) {
-        BSONObjBuilder childBob;
-        statsToBSON(*stats.children[0], verbosity, &childBob, topLevelBob);
-        bob->append("inputStage", childBob.obj());
-        return;
-    }
-
-    // There is more than one child. Recursively call statsToBSON(...) on each
-    // of them and add them to the 'inputStages' array.
-
-    BSONArrayBuilder childrenBob(bob->subarrayStart("inputStages"));
-    for (size_t i = 0; i < stats.children.size(); ++i) {
-        BSONObjBuilder childBob(childrenBob.subobjStart());
-        statsToBSON(*stats.children[i], verbosity, &childBob, topLevelBob);
-    }
-    childrenBob.doneFast();
-}
-
-BSONObj Explain::statsToBSON(const PlanStageStats& stats, ExplainOptions::Verbosity verbosity) {
-    BSONObjBuilder bob;
-    statsToBSON(stats, &bob, verbosity);
-    return bob.obj();
-}
-
-BSONObj Explain::statsToBSON(const sbe::PlanStageStats& stats,
-                             ExplainOptions::Verbosity verbosity) {
-    BSONObjBuilder bob;
-    return bob.obj();
-}
-
-void Explain::statsToBSON(const PlanStageStats& stats,
-                          BSONObjBuilder* bob,
-                          ExplainOptions::Verbosity verbosity) {
-    statsToBSON(stats, verbosity, bob, bob);
-}
-
-void Explain::generatePlannerInfo(PlanExecutor* exec,
-                                  const CollectionPtr& collection,
-                                  BSONObj extraInfo,
-                                  BSONObjBuilder* out) {
-    auto planExecImpl = dynamic_cast<PlanExecutorImpl*>(exec);
-    uassert(4847801,
-            "queryPlanner explain section is only supported for classic PlanStages",
-            planExecImpl);
-
-    CanonicalQuery* query = exec->getCanonicalQuery();
-
+void generatePlannerInfo(PlanExecutor* exec,
+                         const CollectionPtr& collection,
+                         BSONObj extraInfo,
+                         BSONObjBuilder* out) {
     BSONObjBuilder plannerBob(out->subobjStart("queryPlanner"));
 
     plannerBob.append("plannerVersion", QueryPlanner::kPlannerVersion);
     plannerBob.append("namespace", exec->nss().ns());
 
-    // Find whether there is an index filter set for the query shape. The 'indexFilterSet'
-    // field will always be false in the case of EOF or idhack plans.
+    // Find whether there is an index filter set for the query shape. The 'indexFilterSet' field
+    // will always be false in the case of EOF or idhack plans.
     bool indexFilterSet = false;
     boost::optional<uint32_t> queryHash;
     boost::optional<uint32_t> planCacheKeyHash;
@@ -628,9 +107,10 @@ void Explain::generatePlannerInfo(PlanExecutor* exec,
     }
     plannerBob.append("indexFilterSet", indexFilterSet);
 
-    // In general we should have a canonical query, but sometimes we may avoid
-    // creating a canonical query as an optimization (specifically, the update system
-    // does not canonicalize for idhack updates). In these cases, 'query' is NULL.
+    // In general we should have a canonical query, but sometimes we may avoid creating a canonical
+    // query as an optimization (specifically, the update system does not canonicalize for idhack
+    // updates). In these cases, 'query' is NULL.
+    auto query = exec->getCanonicalQuery();
     if (nullptr != query) {
         BSONObjBuilder parsedQueryBob(plannerBob.subobjStart("parsedQuery"));
         query->root()->serialize(&parsedQueryBob);
@@ -653,103 +133,82 @@ void Explain::generatePlannerInfo(PlanExecutor* exec,
         plannerBob.appendElements(extraInfo);
     }
 
-    BSONObjBuilder winningPlanBob(plannerBob.subobjStart("winningPlan"));
-    const auto winnerStats = getWinningPlanStatsTree(planExecImpl);
-    statsToBSON(*winnerStats.get(), &winningPlanBob, ExplainOptions::Verbosity::kQueryPlanner);
-    winningPlanBob.doneFast();
-
-    // Genenerate array of rejected plans.
-    const vector<unique_ptr<PlanStageStats>> rejectedStats =
-        getRejectedPlansTrialStats(planExecImpl);
-    BSONArrayBuilder allPlansBob(plannerBob.subarrayStart("rejectedPlans"));
-    for (size_t i = 0; i < rejectedStats.size(); i++) {
-        BSONObjBuilder childBob(allPlansBob.subobjStart());
-        statsToBSON(*rejectedStats[i], &childBob, ExplainOptions::Verbosity::kQueryPlanner);
-    }
-    allPlansBob.doneFast();
+    auto&& explainer = exec->getPlanExplainer();
+    auto&& [winningStats, _] =
+        explainer.getWinningPlanStats(ExplainOptions::Verbosity::kQueryPlanner);
+    plannerBob.append("winningPlan", winningStats);
 
+    BSONArrayBuilder bab{plannerBob.subarrayStart("rejectedPlans")};
+    for (auto&& [rejectedStats, _] :
+         explainer.getRejectedPlansStats(ExplainOptions::Verbosity::kQueryPlanner)) {
+        bab.append(rejectedStats);
+    }
+    bab.doneFast();
     plannerBob.doneFast();
 }
 
-// static
-void Explain::generateSinglePlanExecutionInfo(const PlanStageStats* stats,
-                                              ExplainOptions::Verbosity verbosity,
-                                              boost::optional<long long> totalTimeMillis,
-                                              BSONObjBuilder* out) {
-    out->appendNumber("nReturned", stats->common.advanced);
+/**
+ * Generates the execution stats section from the given 'PlanStatsDetails', adding the resulting
+ * BSON document and specific execution metrics to 'out'.
+ *
+ * The 'totalTimeMillis' value passed here will be added to the top level of the execution stats
+ * section, but will not affect the reporting of timing for individual stages. If 'totalTimeMillis'
+ * is not set, we use the approximate timing information collected by the stages.
+ *
+ * Stats are generated at the verbosity specified by 'verbosity'.
+ */
+void generateSinglePlanExecutionInfo(const PlanExplainer::PlanStatsDetails& details,
+                                     boost::optional<long long> totalTimeMillis,
+                                     BSONObjBuilder* out) {
+    auto&& [stats, summary] = details;
+    invariant(summary);
+
+    out->appendNumber("nReturned", summary->nReturned);
 
     // Time elapsed could might be either precise or approximate.
     if (totalTimeMillis) {
         out->appendNumber("executionTimeMillis", *totalTimeMillis);
     } else {
-        invariant(stats->common.executionTimeMillis);
-        out->appendNumber("executionTimeMillisEstimate", *stats->common.executionTimeMillis);
+        out->appendNumber("executionTimeMillisEstimate", summary->executionTimeMillisEstimate);
     }
 
-    // Flatten the stats tree into a list.
-    vector<const PlanStageStats*> statsNodes;
-    flattenStatsTree(stats, &statsNodes);
-
-    // Iterate over all stages in the tree and get the total number of keys/docs examined.
-    // These are just aggregations of information already available in the stats tree.
-    size_t totalKeysExamined = 0;
-    size_t totalDocsExamined = 0;
-    for (size_t i = 0; i < statsNodes.size(); ++i) {
-        totalKeysExamined +=
-            getKeysExamined(statsNodes[i]->stageType, statsNodes[i]->specific.get());
-        totalDocsExamined +=
-            getDocsExamined(statsNodes[i]->stageType, statsNodes[i]->specific.get());
-    }
+    out->appendNumber("totalKeysExamined", summary->totalKeysExamined);
+    out->appendNumber("totalDocsExamined", summary->totalDocsExamined);
 
-    out->appendNumber("totalKeysExamined", totalKeysExamined);
-    out->appendNumber("totalDocsExamined", totalDocsExamined);
-    if (stats->common.failed)
-        out->appendBool("failed", stats->common.failed);
+    if (summary->planFailed) {
+        out->appendBool("failed", true);
+    }
 
     // Add the tree of stages, with individual execution stats for each stage.
-    BSONObjBuilder stagesBob(out->subobjStart("executionStages"));
-    statsToBSON(*stats, &stagesBob, verbosity);
-    stagesBob.doneFast();
+    out->append("executionStages", stats);
 }
 
-std::unique_ptr<PlanStageStats> Explain::getWinningPlanTrialStats(PlanExecutor* exec) {
-    auto planExecImpl = dynamic_cast<PlanExecutorImpl*>(exec);
-    uassert(4847802,
-            "getWinningPlanTrialStats() is only supported for classic PlanStages",
-            planExecImpl);
-
-    // Inspect the tree to see if there is a MultiPlanStage. Plan selection has already happened at
-    // this point, since we have a PlanExecutor.
-    const auto mps = planExecImpl->getMultiPlanStage();
-
-    if (mps) {
-        const auto mpsStats = mps->getStats();
-        return std::move(mpsStats->children[mps->bestPlanIdx()]);
-    }
-
-    return nullptr;
-}
+/**
+ * Adds the "executionStats" field to out. Assumes that the PlanExecutor has already been executed
+ * to the point of reaching EOF. Also assumes that verbosity >= kExecStats.
+ *
+ * If verbosity >= kExecAllPlans, it will include the "allPlansExecution" array.
+ *
+ * - 'execPlanStatus' is OK if the query was exected successfully, or a non-OK status if there
+ *   was a runtime error.
+ */
+void generateExecutionInfo(PlanExecutor* exec,
+                           ExplainOptions::Verbosity verbosity,
+                           Status executePlanStatus,
+                           boost::optional<PlanExplainer::PlanStatsDetails> winningPlanTrialStats,
+                           BSONObjBuilder* out) {
+    invariant(verbosity >= ExplainOptions::Verbosity::kExecStats);
 
-void Explain::generateExecutionInfo(PlanExecutor* exec,
-                                    ExplainOptions::Verbosity verbosity,
-                                    Status executePlanStatus,
-                                    PlanStageStats* winningPlanTrialStats,
-                                    BSONObjBuilder* out) {
-    auto planExecImpl = dynamic_cast<PlanExecutorImpl*>(exec);
-    uassert(4847800,
-            "executionStats explain section is only supported for classic PlanStages",
-            planExecImpl);
+    auto&& explainer = exec->getPlanExplainer();
 
-    invariant(verbosity >= ExplainOptions::Verbosity::kExecStats);
-    if (verbosity >= ExplainOptions::Verbosity::kExecAllPlans &&
-        planExecImpl->getMultiPlanStage()) {
+    if (verbosity >= ExplainOptions::Verbosity::kExecAllPlans && explainer.isMultiPlan()) {
         invariant(winningPlanTrialStats,
-                  "winningPlanTrialStats must be non-null when requesting all execution stats");
+                  "winningPlanTrialStats must be present when requesting all execution stats");
     }
     BSONObjBuilder execBob(out->subobjStart("executionStats"));
 
-    // If there is an execution error while running the query, the error is reported under
-    // the "executionStats" section and the explain as a whole succeeds.
+    // If there is an execution error while running the query, the error is reported under the
+    // "executionStats" section and the explain as a whole succeeds.
     execBob.append("executionSuccess", executePlanStatus.isOK());
     if (!executePlanStatus.isOK()) {
         execBob.append("errorMessage", executePlanStatus.reason());
@@ -757,34 +216,29 @@ void Explain::generateExecutionInfo(PlanExecutor* exec,
     }
 
     // Generate exec stats BSON for the winning plan.
-    OperationContext* opCtx = exec->getOpCtx();
-    long long totalTimeMillis = durationCount<Milliseconds>(CurOp::get(opCtx)->elapsedTimeTotal());
-    const auto winningExecStats = getWinningPlanStatsTree(planExecImpl);
-    generateSinglePlanExecutionInfo(winningExecStats.get(), verbosity, totalTimeMillis, &execBob);
+    auto opCtx = exec->getOpCtx();
+    auto totalTimeMillis = durationCount<Milliseconds>(CurOp::get(opCtx)->elapsedTimeTotal());
+    generateSinglePlanExecutionInfo(
+        explainer.getWinningPlanStats(verbosity), totalTimeMillis, &execBob);
 
-    // Also generate exec stats for all plans, if the verbosity level is high enough.
-    // These stats reflect what happened during the trial period that ranked the plans.
+    // Also generate exec stats for all plans, if the verbosity level is high enough. These stats
+    // reflect what happened during the trial period that ranked the plans.
     if (verbosity >= ExplainOptions::Verbosity::kExecAllPlans) {
-        // If we ranked multiple plans against each other, then add stats collected
-        // from the trial period of the winning plan. The "allPlansExecution" section
-        // will contain an apples-to-apples comparison of the winning plan's stats against
-        // all rejected plans' stats collected during the trial period.
-
+        // If we ranked multiple plans against each other, then add stats collected from the trial
+        // period of the winning plan. The "allPlansExecution" section will contain an
+        // apples-to-apples comparison of the winning plan's stats against all rejected plans' stats
+        // collected during the trial period.
         BSONArrayBuilder allPlansBob(execBob.subarrayStart("allPlansExecution"));
 
         if (winningPlanTrialStats) {
             BSONObjBuilder planBob(allPlansBob.subobjStart());
-            generateSinglePlanExecutionInfo(
-                winningPlanTrialStats, verbosity, boost::none, &planBob);
+            generateSinglePlanExecutionInfo(*winningPlanTrialStats, boost::none, &planBob);
             planBob.doneFast();
         }
 
-        const vector<unique_ptr<PlanStageStats>> rejectedStats =
-            getRejectedPlansTrialStats(planExecImpl);
-        for (size_t i = 0; i < rejectedStats.size(); ++i) {
+        for (auto&& stats : explainer.getRejectedPlansStats(verbosity)) {
             BSONObjBuilder planBob(allPlansBob.subobjStart());
-            generateSinglePlanExecutionInfo(
-                rejectedStats[i].get(), verbosity, boost::none, &planBob);
+            generateSinglePlanExecutionInfo(stats, boost::none, &planBob);
             planBob.doneFast();
         }
 
@@ -794,11 +248,25 @@ void Explain::generateExecutionInfo(PlanExecutor* exec,
     execBob.doneFast();
 }
 
+/**
+ * Executes the given plan executor, discarding the resulting documents, until it reaches EOF. If a
+ * runtime error occur or execution is killed, throws a DBException.
+ *
+ * If 'exec' is configured for yielding, then a call to this helper could result in a yield.
+ */
+void executePlan(PlanExecutor* exec) {
+    BSONObj obj;
+    while (exec->getNext(&obj, nullptr) == PlanExecutor::ADVANCED) {
+        // Discard the resulting documents.
+    }
+}
+}  // namespace
+
 void Explain::explainStages(PlanExecutor* exec,
                             const CollectionPtr& collection,
                             ExplainOptions::Verbosity verbosity,
                             Status executePlanStatus,
-                            PlanStageStats* winningPlanTrialStats,
+                            boost::optional<PlanExplainer::PlanStatsDetails> winningPlanTrialStats,
                             BSONObj extraInfo,
                             BSONObjBuilder* out) {
     //
@@ -814,20 +282,24 @@ void Explain::explainStages(PlanExecutor* exec,
     }
 }
 
-void Explain::explainPipelineExecutor(PlanExecutorPipeline* exec,
-                                      ExplainOptions::Verbosity verbosity,
-                                      BSONObjBuilder* out) {
+void Explain::explainPipeline(PlanExecutor* exec,
+                              bool executePipeline,
+                              ExplainOptions::Verbosity verbosity,
+                              BSONObjBuilder* out) {
     invariant(exec);
     invariant(out);
 
+    auto pipelineExec = dynamic_cast<PlanExecutorPipeline*>(exec);
+    invariant(pipelineExec);
+
     // If we need execution stats, this runs the plan in order to gather the stats.
-    if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
+    if (verbosity >= ExplainOptions::Verbosity::kExecStats && executePipeline) {
         // TODO SERVER-32732: An execution error should be reported in explain, but should not
         // cause the explain itself to fail.
-        executePlan(exec);
+        executePlan(pipelineExec);
     }
 
-    *out << "stages" << Value(exec->writeExplainOps(verbosity));
+    *out << "stages" << Value(pipelineExec->writeExplainOps(verbosity));
 
     explain_common::generateServerInfo(out);
 }
@@ -837,12 +309,8 @@ void Explain::explainStages(PlanExecutor* exec,
                             ExplainOptions::Verbosity verbosity,
                             BSONObj extraInfo,
                             BSONObjBuilder* out) {
-    uassert(4822877,
-            "explainStages() is only supported for PlanStage trees",
-            dynamic_cast<PlanExecutorImpl*>(exec));
-
-    auto winningPlanTrialStats = Explain::getWinningPlanTrialStats(exec);
-
+    auto&& explainer = exec->getPlanExplainer();
+    auto winningPlanTrialStats = explainer.getWinningPlanStats(verbosity);
     Status executePlanStatus = Status::OK();
     const CollectionPtr* collectionPtr = &collection;
 
@@ -862,45 +330,12 @@ void Explain::explainStages(PlanExecutor* exec,
         }
     }
 
-    explainStages(exec,
-                  *collectionPtr,
-                  verbosity,
-                  executePlanStatus,
-                  winningPlanTrialStats.get(),
-                  extraInfo,
-                  out);
+    explainStages(
+        exec, *collectionPtr, verbosity, executePlanStatus, winningPlanTrialStats, extraInfo, out);
 
     explain_common::generateServerInfo(out);
 }
 
-std::string Explain::getPlanSummary(const PlanStage* root) {
-    std::vector<const PlanStage*> stages;
-    flattenExecTree(root, &stages);
-
-    // Use this stream to build the plan summary string.
-    StringBuilder sb;
-    bool seenLeaf = false;
-
-    for (size_t i = 0; i < stages.size(); i++) {
-        if (stages[i]->getChildren().empty()) {
-            // This is a leaf node. Add to the plan summary string accordingly. Unless
-            // this is the first leaf we've seen, add a delimiting string first.
-            if (seenLeaf) {
-                sb << ", ";
-            } else {
-                seenLeaf = true;
-            }
-            addStageSummaryStr(stages[i], sb);
-        }
-    }
-
-    return sb.str();
-}
-
-std::string Explain::getPlanSummary(const sbe::PlanStage* root) {
-    // TODO: Support 'planSummary' for SBE plan stage trees.
-    return "unsupported";
-}
 
 void Explain::planCacheEntryToBSON(const PlanCacheEntry& entry, BSONObjBuilder* out) {
     BSONObjBuilder shapeBuilder(out->subobjStart("createdFromQuery"));
@@ -918,19 +353,20 @@ void Explain::planCacheEntryToBSON(const PlanCacheEntry& entry, BSONObjBuilder*
     out->append("isActive", entry.isActive);
     out->append("works", static_cast<long long>(entry.works));
 
-    BSONObjBuilder cachedPlanBob(out->subobjStart("cachedPlan"));
-    Explain::statsToBSON(*(entry.decision->getStats<PlanStageStats>()[0]),
-                         &cachedPlanBob,
-                         ExplainOptions::Verbosity::kQueryPlanner);
-    cachedPlanBob.doneFast();
 
+    auto explainer = plan_explainer_factory::makePlanExplainer<PlanStage>(nullptr, nullptr);
+    auto plannerStats =
+        explainer->getCachedPlanStats(entry, ExplainOptions::Verbosity::kQueryPlanner);
+    auto execStats = explainer->getCachedPlanStats(entry, ExplainOptions::Verbosity::kExecStats);
+
+    invariant(plannerStats.size() > 0);
+    out->append("cachedPlan", plannerStats[0].first);
     out->append("timeOfCreation", entry.timeOfCreation);
 
     BSONArrayBuilder creationBuilder(out->subarrayStart("creationExecStats"));
-    for (auto&& stat : entry.decision->getStats<PlanStageStats>()) {
+    for (auto&& stats : execStats) {
         BSONObjBuilder planBob(creationBuilder.subobjStart());
-        Explain::generateSinglePlanExecutionInfo(
-            stat.get(), ExplainOptions::Verbosity::kExecAllPlans, boost::none, &planBob);
+        generateSinglePlanExecutionInfo(stats, boost::none, &planBob);
         planBob.doneFast();
     }
     creationBuilder.doneFast();
@@ -948,5 +384,4 @@ void Explain::planCacheEntryToBSON(const PlanCacheEntry& entry, BSONObjBuilder*
 
     out->append("indexFilterSet", entry.plannerData[0]->indexFilterApplied);
 }
-
 }  // namespace mongo
diff --git a/src/mongo/db/query/explain.h b/src/mongo/db/query/explain.h
index a8848decfc1..b59cccb1c96 100644
--- a/src/mongo/db/query/explain.h
+++ b/src/mongo/db/query/explain.h
@@ -29,15 +29,10 @@
 
 #pragma once
 
-#include "mongo/db/exec/plan_stage.h"
-#include "mongo/db/exec/plan_stats.h"
-#include "mongo/db/exec/sbe/stages/plan_stats.h"
-#include "mongo/db/exec/sbe/stages/stages.h"
-#include "mongo/db/query/canonical_query.h"
 #include "mongo/db/query/explain_options.h"
+#include "mongo/db/query/plan_cache.h"
 #include "mongo/db/query/plan_executor.h"
-#include "mongo/db/query/query_planner_params.h"
-#include "mongo/db/query/query_solution.h"
+#include "mongo/db/query/plan_explainer.h"
 
 namespace mongo {
 
@@ -66,8 +61,9 @@ public:
      *
      * Does not take ownership of its arguments.
      *
-     * The caller should hold at least an IS lock on the collection the that the query runs on,
-     * even if 'collection' is nullptr.
+     * During this call it may be required to execute the plan to collect statistics. If the
+     * PlanExecutor uses 'kLockExternally' lock policy, the caller should hold at least an IS lock
+     * on the collection the that the query runs on, even if 'collection' parameter is nullptr.
      *
      * If there is an error during the execution of the query, the error message and code are
      * added to the "executionStats" section of the explain.
@@ -82,8 +78,10 @@ public:
      * the other overload of explainStages() above, this one does not add the "serverInfo" section.
      *
      * - 'exec' is the stage tree for the operation being explained.
-     * - 'collection' is the relevant collection. The caller should hold at least an IS lock on the
-     * collection which the query ran on, even 'collection' is nullptr.
+     * - 'collection' is the relevant collection. During this call it may be required to execute the
+     * plan to collect statistics. If the PlanExecutor uses 'kLockExternally' lock policy, the
+     * caller should hold at least an IS lock on the collection the that the query runs on, even if
+     * 'collection' parameter is nullptr.
      * - 'verbosity' is the verbosity level of the explain.
      * - 'extraInfo' specifies additional information to include into the output.
      * - 'executePlanStatus' is the status returned after executing the query (Status::OK if the
@@ -92,64 +90,29 @@ public:
      * nullptr.
      * - 'out' is the builder for the explain output.
      */
-    static void explainStages(PlanExecutor* exec,
-                              const CollectionPtr& collection,
-                              ExplainOptions::Verbosity verbosity,
-                              Status executePlanStatus,
-                              PlanStageStats* winningPlanTrialStats,
-                              BSONObj extraInfo,
-                              BSONObjBuilder* out);
+    static void explainStages(
+        PlanExecutor* exec,
+        const CollectionPtr& collection,
+        ExplainOptions::Verbosity verbosity,
+        Status executePlanStatus,
+        boost::optional<PlanExplainer::PlanStatsDetails> winningPlanTrialStats,
+        BSONObj extraInfo,
+        BSONObjBuilder* out);
 
     /**
      * Gets explain BSON for the document sources contained by 'exec'. Use this function if you have
      * a PlanExecutor for a pipeline and want to turn it into a human readable explain format.
      *
      * The explain information is generated with the level of detail specified by 'verbosity'.
-     */
-    static void explainPipelineExecutor(PlanExecutorPipeline* exec,
-                                        ExplainOptions::Verbosity verbosity,
-                                        BSONObjBuilder* out);
-
-    /**
-     * Converts the stats tree 'stats' into a corresponding BSON object containing
-     * explain information.
-     *
-     * Generates the BSON stats at a verbosity specified by 'verbosity'. Defaults
-     * to execution stats verbosity.
-     */
-    static BSONObj statsToBSON(
-        const PlanStageStats& stats,
-        ExplainOptions::Verbosity verbosity = ExplainOptions::Verbosity::kExecStats);
-    static BSONObj statsToBSON(
-        const sbe::PlanStageStats& stats,
-        ExplainOptions::Verbosity verbosity = ExplainOptions::Verbosity::kExecStats);
-
-    /**
-     * This version of stats tree to BSON conversion returns the result through the
-     * out-parameter 'bob' rather than returning a BSONObj.
      *
-     * Generates the BSON stats at a verbosity specified by 'verbosity'. Defaults
-     * to execution stats verbosity.
+     * If 'verbosity' >= 'kExecStats' the 'executePipeline' flag is used to indicate whether the
+     * pipeline needs to be executed first, before the stats is collected. Otherwise, it is assumed
+     * that the plan was already executed until EOF and the stats are ready for collection.
      */
-    static void statsToBSON(
-        const PlanStageStats& stats,
-        BSONObjBuilder* bob,
-        ExplainOptions::Verbosity verbosity = ExplainOptions::Verbosity::kExecStats);
-
-    /**
-     * Returns a short plan summary std::string describing the leaves of the query plan.
-     */
-    static std::string getPlanSummary(const PlanStage* root);
-    static std::string getPlanSummary(const sbe::PlanStage* root);
-
-    /**
-     * If exec's root stage is a MultiPlanStage, returns the stats for the trial period of of the
-     * winning plan. Otherwise, returns nullptr.
-     *
-     * Must be called _before_ executing the plan with PlanExecutor::getNext()
-     * or the PlanExecutor::execute*() methods.
-     */
-    static std::unique_ptr<PlanStageStats> getWinningPlanTrialStats(PlanExecutor* exec);
+    static void explainPipeline(PlanExecutor* exec,
+                                bool executePipeline,
+                                ExplainOptions::Verbosity verbosity,
+                                BSONObjBuilder* out);
 
     /**
      * Serializes a PlanCacheEntry to the provided BSON object builder. The output format is
@@ -157,85 +120,6 @@ public:
      * the plan cache.
      */
     static void planCacheEntryToBSON(const PlanCacheEntry& entry, BSONObjBuilder* out);
-
-    /**
-     * Traverses the tree rooted at 'root', and adds all nodes into the list 'flattened'. If a
-     * MultiPlanStage is encountered, only adds the best plan and its children to 'flattened'.
-     */
-    static void flattenExecTree(const PlanStage* root, std::vector<const PlanStage*>* flattened);
-
-    /**
-     * Given the SpecificStats object for a stage and the type of the stage, returns the number of
-     * index keys examined by the stage.
-     */
-    static size_t getKeysExamined(StageType type, const SpecificStats* specific);
-
-    /**
-     * Given the SpecificStats object for a stage and the type of the stage, returns the number of
-     * documents examined by the stage.
-     */
-    static size_t getDocsExamined(StageType type, const SpecificStats* specific);
-
-private:
-    /**
-     * Generates the execution stats section for the stats tree 'stats', adding the resulting BSON
-     * to 'out'.
-     *
-     * The 'totalTimeMillis' value passed here will be added to the top level of the execution stats
-     * section, but will not affect the reporting of timing for individual stages. If
-     * 'totalTimeMillis' is not set, we use the approximate timing information collected by the
-     * stages.
-     *
-     * Stats are generated at the verbosity specified by 'verbosity'.
-     **/
-    static void generateSinglePlanExecutionInfo(const PlanStageStats* stats,
-                                                ExplainOptions::Verbosity verbosity,
-                                                boost::optional<long long> totalTimeMillis,
-                                                BSONObjBuilder* out);
-
-    /**
-     * Adds the 'queryPlanner' explain section to the BSON object being built
-     * by 'out'.
-     *
-     * This is a helper for generating explain BSON. It is used by explainStages(...).
-     *
-     * - 'exec' is the stage tree for the operation being explained.
-     * - 'collection' is the collection used in the operation. The caller should hold an IS lock on
-     * the collection which the query is for, even if 'collection' is nullptr.
-     * - 'extraInfo' specifies additional information to include into the output.
-     * - 'out' is a builder for the explain output.
-     */
-    static void generatePlannerInfo(PlanExecutor* exec,
-                                    const CollectionPtr& collection,
-                                    BSONObj extraInfo,
-                                    BSONObjBuilder* out);
-
-    /**
-     * Private helper that does the heavy-lifting for the public statsToBSON(...) functions
-     * declared above.
-     *
-     * Not used except as a helper to the public statsToBSON(...) functions.
-     */
-    static void statsToBSON(const PlanStageStats& stats,
-                            ExplainOptions::Verbosity verbosity,
-                            BSONObjBuilder* bob,
-                            BSONObjBuilder* topLevelBob);
-
-    /**
-     * Adds the "executionStats" field to out. Assumes that the PlanExecutor has already been
-     * executed to the point of reaching EOF. Also assumes that verbosity >= kExecStats.
-     *
-     * If verbosity >= kExecAllPlans, it will include the "allPlansExecution" array.
-     *
-     * - 'execPlanStatus' is OK if the query was exected successfully, or a non-OK status if there
-     *   was a runtime error.
-     * - 'winningPlanTrialStats' may be nullptr.
-     **/
-    static void generateExecutionInfo(PlanExecutor* exec,
-                                      ExplainOptions::Verbosity verbosity,
-                                      Status executePlanStatus,
-                                      PlanStageStats* winningPlanTrialStats,
-                                      BSONObjBuilder* out);
 };
 
 }  // namespace mongo
diff --git a/src/mongo/db/query/find.cpp b/src/mongo/db/query/find.cpp
index 27e10256727..b980c9a9ef6 100644
--- a/src/mongo/db/query/find.cpp
+++ b/src/mongo/db/query/find.cpp
@@ -134,7 +134,8 @@ void endQueryOp(OperationContext* opCtx,
 
     // Fill out CurOp based on explain summary statistics.
     PlanSummaryStats summaryStats;
-    exec.getSummaryStats(&summaryStats);
+    auto&& explainer = exec.getPlanExplainer();
+    explainer.getSummaryStats(&summaryStats);
     curOp->debug().setPlanSummaryMetrics(summaryStats);
 
     if (collection) {
@@ -142,7 +143,8 @@ void endQueryOp(OperationContext* opCtx,
     }
 
     if (curOp->shouldDBProfile(opCtx)) {
-        curOp->debug().execStats = exec.getStats();
+        auto&& [stats, _] = explainer.getWinningPlanStats(ExplainOptions::Verbosity::kExecStats);
+        curOp->debug().execStats = std::move(stats);
     }
 }
 
@@ -181,7 +183,9 @@ void generateBatch(int ntoreturn,
             (*numResults)++;
         }
     } catch (DBException& exception) {
-        LOGV2_ERROR(20918, "getMore executor error", "stats"_attr = redact(exec->getStats()));
+        auto&& explainer = exec->getPlanExplainer();
+        auto&& [stats, _] = explainer.getWinningPlanStats(ExplainOptions::Verbosity::kExecStats);
+        LOGV2_ERROR(20918, "getMore executor error", "stats"_attr = redact(stats));
         exception.addContext("Executor error during OP_GET_MORE");
         throw;
     }
@@ -412,7 +416,7 @@ Message getMore(OperationContext* opCtx,
     exec->reattachToOperationContext(opCtx);
     exec->restoreState(readLock ? &readLock->getCollection() : nullptr);
 
-    auto planSummary = exec->getPlanSummary();
+    auto planSummary = exec->getPlanExplainer().getPlanSummary();
     {
         stdx::lock_guard<Client> lk(*opCtx->getClient());
         curOp.setPlanSummary_inlock(planSummary);
@@ -448,7 +452,7 @@ Message getMore(OperationContext* opCtx,
     // these values we need to take a diff of the pre-execution and post-execution metrics, as they
     // accumulate over the course of a cursor's lifetime.
     PlanSummaryStats preExecutionStats;
-    exec->getSummaryStats(&preExecutionStats);
+    exec->getPlanExplainer().getSummaryStats(&preExecutionStats);
     if (MONGO_unlikely(waitWithPinnedCursorDuringGetMoreBatch.shouldFail())) {
         CurOpFailpointHelpers::waitWhileFailPointEnabled(&waitWithPinnedCursorDuringGetMoreBatch,
                                                          opCtx,
@@ -484,7 +488,8 @@ Message getMore(OperationContext* opCtx,
     }
 
     PlanSummaryStats postExecutionStats;
-    exec->getSummaryStats(&postExecutionStats);
+    auto&& explainer = exec->getPlanExplainer();
+    explainer.getSummaryStats(&postExecutionStats);
     postExecutionStats.totalKeysExamined -= preExecutionStats.totalKeysExamined;
     postExecutionStats.totalDocsExamined -= preExecutionStats.totalDocsExamined;
     curOp.debug().setPlanSummaryMetrics(postExecutionStats);
@@ -496,7 +501,9 @@ Message getMore(OperationContext* opCtx,
     // cost.
     if (cursorPin->getExecutor()->lockPolicy() != PlanExecutor::LockPolicy::kLocksInternally &&
         curOp.shouldDBProfile(opCtx)) {
-        curOp.debug().execStats = exec->getStats();
+        auto&& [stats, _] = explainer.getWinningPlanStats(ExplainOptions::Verbosity::kExecStats);
+
+        curOp.debug().execStats = std::move(stats);
     }
 
     // Our two possible ClientCursorPin cleanup paths are:
@@ -682,7 +689,7 @@ bool runQuery(OperationContext* opCtx,
     // Get summary info about which plan the executor is using.
     {
         stdx::lock_guard<Client> lk(*opCtx->getClient());
-        curOp.setPlanSummary_inlock(exec->getPlanSummary());
+        curOp.setPlanSummary_inlock(exec->getPlanExplainer().getPlanSummary());
     }
 
     try {
@@ -710,10 +717,12 @@ bool runQuery(OperationContext* opCtx,
             }
         }
     } catch (DBException& exception) {
+        auto&& explainer = exec->getPlanExplainer();
+        auto&& [stats, _] = explainer.getWinningPlanStats(ExplainOptions::Verbosity::kExecStats);
         LOGV2_ERROR(20919,
                     "Plan executor error during find",
                     "error"_attr = redact(exception.toStatus()),
-                    "stats"_attr = redact(exec->getStats()));
+                    "stats"_attr = redact(stats));
 
         exception.addContext("Executor error during find");
         throw;
diff --git a/src/mongo/db/query/get_executor.cpp b/src/mongo/db/query/get_executor.cpp
index daecec36c15..d5e64f2fd05 100644
--- a/src/mongo/db/query/get_executor.cpp
+++ b/src/mongo/db/query/get_executor.cpp
@@ -415,7 +415,8 @@ public:
 
     std::string getPlanSummary() const final {
         invariant(_root);
-        return Explain::getPlanSummary(_root.get());
+        auto explainer = plan_explainer_factory::makePlanExplainer(_root.get(), nullptr);
+        return explainer->getPlanSummary();
     }
 
     std::unique_ptr<PlanStage> root() {
@@ -466,8 +467,11 @@ public:
     std::string getPlanSummary() const final {
         // We can report plan summary only if this result contains a single solution.
         invariant(_roots.size() == 1);
+        invariant(_solutions.size() == 1);
         invariant(_roots[0].first);
-        return Explain::getPlanSummary(_roots[0].first.get());
+        auto explainer =
+            plan_explainer_factory::makePlanExplainer(_roots[0].first.get(), _solutions[0].get());
+        return explainer->getPlanSummary();
     }
 
     PlanStageVector roots() {
@@ -2057,11 +2061,12 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorForS
     auto&& root = stage_builder::buildClassicExecutableTree(
         opCtx, collection, *parsedDistinct->getQuery(), *soln, ws.get());
 
+    auto explainer = plan_explainer_factory::makePlanExplainer(root.get(), soln.get());
     LOGV2_DEBUG(20931,
                 2,
                 "Using fast distinct",
                 "query"_attr = redact(parsedDistinct->getQuery()->toStringShort()),
-                "planSummary"_attr = Explain::getPlanSummary(root.get()));
+                "planSummary"_attr = explainer->getPlanSummary());
 
     return plan_executor_factory::make(parsedDistinct->releaseQuery(),
                                        std::move(ws),
@@ -2101,11 +2106,13 @@ getExecutorDistinctFromIndexSolutions(OperationContext* opCtx,
             auto&& root = stage_builder::buildClassicExecutableTree(
                 opCtx, collection, *parsedDistinct->getQuery(), *currentSolution, ws.get());
 
+            auto explainer =
+                plan_explainer_factory::makePlanExplainer(root.get(), currentSolution.get());
             LOGV2_DEBUG(20932,
                         2,
                         "Using fast distinct",
                         "query"_attr = redact(parsedDistinct->getQuery()->toStringShort()),
-                        "planSummary"_attr = Explain::getPlanSummary(root.get()));
+                        "planSummary"_attr = explainer->getPlanSummary());
 
             return plan_executor_factory::make(parsedDistinct->releaseQuery(),
                                                std::move(ws),
diff --git a/src/mongo/db/query/plan_executor.h b/src/mongo/db/query/plan_executor.h
index 60a832e0491..0370c46b5da 100644
--- a/src/mongo/db/query/plan_executor.h
+++ b/src/mongo/db/query/plan_executor.h
@@ -34,6 +34,7 @@
 #include "mongo/db/operation_context.h"
 #include "mongo/db/ops/update_result.h"
 #include "mongo/db/query/canonical_query.h"
+#include "mongo/db/query/plan_explainer.h"
 #include "mongo/db/query/plan_summary_stats.h"
 #include "mongo/db/query/plan_yield_policy.h"
 
@@ -328,30 +329,14 @@ public:
     virtual LockPolicy lockPolicy() const = 0;
 
     /**
-     * Returns a short string, suitable for the logs, which summarizes the execution plan.
-     */
-    virtual std::string getPlanSummary() const = 0;
-
-    /**
-     * Fills out 'statsOut' with summary stats collected during the execution of the PlanExecutor.
-     * This is a lightweight alternative which is useful when operations want to request a summary
-     * of the available debug information without generating complete explain output.
-     *
-     * The summary stats are consumed by debug mechanisms such as the profiler and the slow query
-     * log.
-     */
-    virtual void getSummaryStats(PlanSummaryStats* statsOut) const = 0;
-
-    /**
-     * Serializes any execution stats tracked by this executor to BSON, for debugging. The format of
-     * these stats are opaque to the caller, and different implementations may choose to provide
-     * different stats.
+     * Returns a PlanExplainer instance to generate plan details and execution stats tracked by this
+     * executor.
      *
      * Implementations must be able to successfully generate and return stats even if the
      * PlanExecutor has issued a query-fatal exception and the executor cannot be used for further
      * query execution.
      */
-    virtual BSONObj getStats() const = 0;
+    virtual const PlanExplainer& getPlanExplainer() const = 0;
 };
 
 }  // namespace mongo
diff --git a/src/mongo/db/query/plan_executor_impl.cpp b/src/mongo/db/query/plan_executor_impl.cpp
index 915bc2ad34b..bb32e0aa566 100644
--- a/src/mongo/db/query/plan_executor_impl.cpp
+++ b/src/mongo/db/query/plan_executor_impl.cpp
@@ -55,6 +55,7 @@
 #include "mongo/db/exec/working_set.h"
 #include "mongo/db/exec/working_set_common.h"
 #include "mongo/db/query/mock_yield_policies.h"
+#include "mongo/db/query/plan_explainer_factory.h"
 #include "mongo/db/query/plan_insert_listener.h"
 #include "mongo/db/query/plan_yield_policy_impl.h"
 #include "mongo/db/repl/replication_coordinator.h"
@@ -103,26 +104,6 @@ std::unique_ptr<PlanYieldPolicy> makeYieldPolicy(PlanExecutorImpl* exec,
             MONGO_UNREACHABLE;
     }
 }
-
-/**
- * Retrieves the first stage of a given type from the plan tree, or NULL
- * if no such stage is found.
- */
-PlanStage* getStageByType(PlanStage* root, StageType type) {
-    if (root->stageType() == type) {
-        return root;
-    }
-
-    const auto& children = root->getChildren();
-    for (size_t i = 0; i < children.size(); i++) {
-        PlanStage* result = getStageByType(children[i].get(), type);
-        if (result) {
-            return result;
-        }
-    }
-
-    return nullptr;
-}
 }  // namespace
 
 PlanExecutorImpl::PlanExecutorImpl(OperationContext* opCtx,
@@ -140,6 +121,7 @@ PlanExecutorImpl::PlanExecutorImpl(OperationContext* opCtx,
       _workingSet(std::move(ws)),
       _qs(std::move(qs)),
       _root(std::move(rt)),
+      _planExplainer(plan_explainer_factory::makePlanExplainer(_root.get(), nullptr)),
       _nss(std::move(nss)),
       // There's no point in yielding if the collection doesn't exist.
       _yieldPolicy(
@@ -607,95 +589,9 @@ PlanExecutor::LockPolicy PlanExecutorImpl::lockPolicy() const {
     return LockPolicy::kLockExternally;
 }
 
-std::string PlanExecutorImpl::getPlanSummary() const {
-    return Explain::getPlanSummary(_root.get());
-}
-
-void PlanExecutorImpl::getSummaryStats(PlanSummaryStats* statsOut) const {
-    invariant(statsOut);
-
-    // We can get some of the fields we need from the common stats stored in the
-    // root stage of the plan tree.
-    const CommonStats* common = _root->getCommonStats();
-    statsOut->nReturned = common->advanced;
-
-    // The other fields are aggregations over the stages in the plan tree. We flatten
-    // the tree into a list and then compute these aggregations.
-    std::vector<const PlanStage*> stages;
-    Explain::flattenExecTree(_root.get(), &stages);
-
-    statsOut->totalKeysExamined = 0;
-    statsOut->totalDocsExamined = 0;
-
-    for (size_t i = 0; i < stages.size(); i++) {
-        statsOut->totalKeysExamined +=
-            Explain::getKeysExamined(stages[i]->stageType(), stages[i]->getSpecificStats());
-        statsOut->totalDocsExamined +=
-            Explain::getDocsExamined(stages[i]->stageType(), stages[i]->getSpecificStats());
-
-        if (isSortStageType(stages[i]->stageType())) {
-            statsOut->hasSortStage = true;
-
-            auto sortStage = static_cast<const SortStage*>(stages[i]);
-            auto sortStats = static_cast<const SortStats*>(sortStage->getSpecificStats());
-            statsOut->usedDisk = sortStats->wasDiskUsed;
-        }
-
-        if (STAGE_IXSCAN == stages[i]->stageType()) {
-            const IndexScan* ixscan = static_cast<const IndexScan*>(stages[i]);
-            const IndexScanStats* ixscanStats =
-                static_cast<const IndexScanStats*>(ixscan->getSpecificStats());
-            statsOut->indexesUsed.insert(ixscanStats->indexName);
-        } else if (STAGE_COUNT_SCAN == stages[i]->stageType()) {
-            const CountScan* countScan = static_cast<const CountScan*>(stages[i]);
-            const CountScanStats* countScanStats =
-                static_cast<const CountScanStats*>(countScan->getSpecificStats());
-            statsOut->indexesUsed.insert(countScanStats->indexName);
-        } else if (STAGE_IDHACK == stages[i]->stageType()) {
-            const IDHackStage* idHackStage = static_cast<const IDHackStage*>(stages[i]);
-            const IDHackStats* idHackStats =
-                static_cast<const IDHackStats*>(idHackStage->getSpecificStats());
-            statsOut->indexesUsed.insert(idHackStats->indexName);
-        } else if (STAGE_DISTINCT_SCAN == stages[i]->stageType()) {
-            const DistinctScan* distinctScan = static_cast<const DistinctScan*>(stages[i]);
-            const DistinctScanStats* distinctScanStats =
-                static_cast<const DistinctScanStats*>(distinctScan->getSpecificStats());
-            statsOut->indexesUsed.insert(distinctScanStats->indexName);
-        } else if (STAGE_TEXT == stages[i]->stageType()) {
-            const TextStage* textStage = static_cast<const TextStage*>(stages[i]);
-            const TextStats* textStats =
-                static_cast<const TextStats*>(textStage->getSpecificStats());
-            statsOut->indexesUsed.insert(textStats->indexName);
-        } else if (STAGE_GEO_NEAR_2D == stages[i]->stageType() ||
-                   STAGE_GEO_NEAR_2DSPHERE == stages[i]->stageType()) {
-            const NearStage* nearStage = static_cast<const NearStage*>(stages[i]);
-            const NearStats* nearStats =
-                static_cast<const NearStats*>(nearStage->getSpecificStats());
-            statsOut->indexesUsed.insert(nearStats->indexName);
-        } else if (STAGE_CACHED_PLAN == stages[i]->stageType()) {
-            const CachedPlanStage* cachedPlan = static_cast<const CachedPlanStage*>(stages[i]);
-            const CachedPlanStats* cachedStats =
-                static_cast<const CachedPlanStats*>(cachedPlan->getSpecificStats());
-            statsOut->replanReason = cachedStats->replanReason;
-        } else if (STAGE_MULTI_PLAN == stages[i]->stageType()) {
-            statsOut->fromMultiPlanner = true;
-        } else if (STAGE_COLLSCAN == stages[i]->stageType()) {
-            statsOut->collectionScans++;
-            const auto collScan = static_cast<const CollectionScan*>(stages[i]);
-            const auto collScanStats =
-                static_cast<const CollectionScanStats*>(collScan->getSpecificStats());
-            if (!collScanStats->tailable)
-                statsOut->collectionScansNonTailable++;
-        }
-    }
-}
-
-BSONObj PlanExecutorImpl::getStats() const {
-    // Serialize all stats from the winning plan.
-    auto mps = getMultiPlanStage();
-    auto winningPlanStats =
-        mps ? std::move(mps->getStats()->children[mps->bestPlanIdx()]) : _root->getStats();
-    return Explain::statsToBSON(*winningPlanStats);
+const PlanExplainer& PlanExecutorImpl::getPlanExplainer() const {
+    invariant(_planExplainer);
+    return *_planExplainer;
 }
 
 MultiPlanStage* PlanExecutorImpl::getMultiPlanStage() const {
diff --git a/src/mongo/db/query/plan_executor_impl.h b/src/mongo/db/query/plan_executor_impl.h
index 651937eeb3f..61eae2bf06f 100644
--- a/src/mongo/db/query/plan_executor_impl.h
+++ b/src/mongo/db/query/plan_executor_impl.h
@@ -87,9 +87,7 @@ public:
     Timestamp getLatestOplogTimestamp() const final;
     BSONObj getPostBatchResumeToken() const final;
     LockPolicy lockPolicy() const final;
-    std::string getPlanSummary() const final;
-    void getSummaryStats(PlanSummaryStats* statsOut) const final;
-    BSONObj getStats() const final;
+    const PlanExplainer& getPlanExplainer() const final;
 
     /**
      * Same as restoreState() but without the logic to retry if a WriteConflictException is thrown.
@@ -148,6 +146,7 @@ private:
     std::unique_ptr<WorkingSet> _workingSet;
     std::unique_ptr<QuerySolution> _qs;
     std::unique_ptr<PlanStage> _root;
+    std::unique_ptr<PlanExplainer> _planExplainer;
 
     // If _killStatus has a non-OK value, then we have been killed and the value represents the
     // reason for the kill.
diff --git a/src/mongo/db/query/plan_executor_sbe.cpp b/src/mongo/db/query/plan_executor_sbe.cpp
index 4e8931a00f1..9e282c40da6 100644
--- a/src/mongo/db/query/plan_executor_sbe.cpp
+++ b/src/mongo/db/query/plan_executor_sbe.cpp
@@ -34,6 +34,7 @@
 #include "mongo/db/db_raii.h"
 #include "mongo/db/exec/sbe/expressions/expression.h"
 #include "mongo/db/exec/sbe/values/bson.h"
+#include "mongo/db/query/plan_explainer_factory.h"
 #include "mongo/db/query/plan_insert_listener.h"
 #include "mongo/db/query/sbe_stage_builder.h"
 
@@ -54,7 +55,9 @@ PlanExecutorSBE::PlanExecutorSBE(
       _ctx(std::move(root.second.ctx)),
       _root(std::move(root.first)),
       _cq{std::move(cq)},
-      _yieldPolicy(std::move(yieldPolicy)) {
+      _yieldPolicy(std::move(yieldPolicy)),
+      // TODO SERVER-50743: plumb through QuerySolution to init the PlanExplainer.
+      _planExplainer{plan_explainer_factory::makePlanExplainer(_root.get(), nullptr)} {
     invariant(_root);
     invariant(!_nss.isEmpty());
 
diff --git a/src/mongo/db/query/plan_executor_sbe.h b/src/mongo/db/query/plan_executor_sbe.h
index 0e3ebb3505c..44cb51000c1 100644
--- a/src/mongo/db/query/plan_executor_sbe.h
+++ b/src/mongo/db/query/plan_executor_sbe.h
@@ -119,17 +119,9 @@ public:
         return LockPolicy::kLocksInternally;
     }
 
-    // TODO: Support 'planSummary' for SBE.
-    std::string getPlanSummary() const override {
-        return "unsupported";
-    }
-
-    // TODO: Support collection of plan summary stats for SBE.
-    void getSummaryStats(PlanSummaryStats* statsOut) const override {}
-
-    // TODO: Support debug stats for SBE.
-    BSONObj getStats() const override {
-        return BSONObj{};
+    const PlanExplainer& getPlanExplainer() const final {
+        invariant(_planExplainer);
+        return *_planExplainer;
     }
 
 private:
@@ -162,6 +154,8 @@ private:
     std::unique_ptr<CanonicalQuery> _cq;
 
     std::unique_ptr<PlanYieldPolicySBE> _yieldPolicy;
+
+    std::unique_ptr<PlanExplainer> _planExplainer;
 };
 
 /**
diff --git a/src/mongo/db/query/plan_explainer.h b/src/mongo/db/query/plan_explainer.h
new file mode 100644
index 00000000000..f69a438c01f
--- /dev/null
+++ b/src/mongo/db/query/plan_explainer.h
@@ -0,0 +1,101 @@
+/**
+ *    Copyright (C) 2020-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include "mongo/db/exec/plan_stats.h"
+#include "mongo/db/query/explain_options.h"
+#include "mongo/db/query/plan_cache.h"
+#include "mongo/db/query/plan_summary_stats.h"
+
+namespace mongo {
+/**
+ * This interface defines an API to provide information on the execution plans generated by the
+ * query planner for a user query in various formats.
+ */
+class PlanExplainer {
+public:
+    /**
+     * This pair holds a serialized BSON document that details the plan selected by the query
+     * planner, and optional summary stats for an execution tree if the verbosity level for the
+     * generated stats is 'executionStats' or higher. The format of these stats are opaque to the
+     * caller, and different implementations may choose to provide different stats.
+     */
+    using PlanStatsDetails = std::pair<BSONObj, boost::optional<PlanSummaryStats>>;
+
+    virtual ~PlanExplainer() = default;
+
+    /**
+     * Returns 'true' if this PlanExplainer can provide information on the winning plan and rejected
+     * candidate plans, meaning that the QueryPlanner generated multiple candidate plans and the
+     * winning plan was chosen by the multi-planner.
+     */
+    virtual bool isMultiPlan() const = 0;
+
+    /**
+     * Returns a short string, suitable for the logs, which summarizes the execution plan.
+     */
+    virtual std::string getPlanSummary() const = 0;
+
+    /**
+     * Fills out 'statsOut' with summary stats collected during the execution of the underlying
+     * plan. This is a lightweight alternative which is useful when operations want to request a
+     * summary of the available debug information without generating complete explain output.
+     *
+     * The summary stats are consumed by debug mechanisms such as the profiler and the slow query
+     * log.
+     */
+    virtual void getSummaryStats(PlanSummaryStats* statsOut) const = 0;
+
+    /**
+     * Returns statistics that detail the winning plan selected by the multi-planner, or, if no
+     * multi-planning has been performed, for the single plan selected by the QueryPlanner.
+     *
+     * The 'verbosity' level parameter determines the amount of information to be returned.
+     */
+    virtual PlanStatsDetails getWinningPlanStats(ExplainOptions::Verbosity verbosity) const = 0;
+
+    /**
+     * Returns statistics that detail candidate plans rejected by the multi-planner. If no
+     * multi-planning has been performed, an empty vector is returned.
+     *
+     * The 'verbosity' level parameter determines the amount of information to be returned.
+     */
+    virtual std::vector<PlanStatsDetails> getRejectedPlansStats(
+        ExplainOptions::Verbosity verbosity) const = 0;
+
+    /**
+     * Serializes a PlanCacheEntry into the provided BSONObjBuilder. The output format is intended
+     * to be human readable, and useful for debugging query performance problems related to the
+     * plan cache.
+     */
+    virtual std::vector<PlanStatsDetails> getCachedPlanStats(
+        const PlanCacheEntry& entry, ExplainOptions::Verbosity verbosity) const = 0;
+};
+}  // namespace mongo
diff --git a/src/mongo/db/query/plan_explainer_factory.h b/src/mongo/db/query/plan_explainer_factory.h
new file mode 100644
index 00000000000..12a0967f276
--- /dev/null
+++ b/src/mongo/db/query/plan_explainer_factory.h
@@ -0,0 +1,45 @@
+/**
+ *    Copyright (C) 2020-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include "mongo/db/query/plan_explainer_impl.h"
+#include "mongo/db/query/plan_explainer_sbe.h"
+
+namespace mongo::plan_explainer_factory {
+template <typename T>
+std::unique_ptr<PlanExplainer> makePlanExplainer(T* root, const QuerySolution* solution) {
+    if constexpr (std::is_same_v<T, sbe::PlanStage>) {
+        return std::make_unique<PlanExplainerSBE>(root, solution);
+    } else {
+        static_assert(std::is_same_v<T, PlanStage>);
+        return std::make_unique<PlanExplainerImpl>(root);
+    }
+}
+}  // namespace mongo::plan_explainer_factory
diff --git a/src/mongo/db/query/plan_explainer_impl.cpp b/src/mongo/db/query/plan_explainer_impl.cpp
new file mode 100644
index 00000000000..637fd7f254a
--- /dev/null
+++ b/src/mongo/db/query/plan_explainer_impl.cpp
@@ -0,0 +1,734 @@
+/**
+ *    Copyright (C) 2020-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kQuery
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/db/query/plan_explainer_impl.h"
+
+#include "mongo/db/exec/cached_plan.h"
+#include "mongo/db/exec/collection_scan.h"
+#include "mongo/db/exec/count_scan.h"
+#include "mongo/db/exec/distinct_scan.h"
+#include "mongo/db/exec/idhack.h"
+#include "mongo/db/exec/index_scan.h"
+#include "mongo/db/exec/multi_plan.h"
+#include "mongo/db/exec/near.h"
+#include "mongo/db/exec/plan_stage.h"
+#include "mongo/db/exec/plan_stats.h"
+#include "mongo/db/exec/sort.h"
+#include "mongo/db/exec/subplan.h"
+#include "mongo/db/exec/text.h"
+#include "mongo/db/exec/trial_stage.h"
+#include "mongo/db/keypattern.h"
+#include "mongo/db/query/explain.h"
+#include "mongo/util/str.h"
+
+namespace mongo {
+namespace {
+/**
+ * Adds to the plan summary string being built by 'sb' for the execution stage 'stage'.
+ */
+void addStageSummaryStr(const PlanStage* stage, StringBuilder& sb) {
+    // First add the stage type string.
+    const CommonStats* common = stage->getCommonStats();
+    sb << common->stageTypeStr;
+
+    // Some leaf nodes also provide info about the index they used.
+    const SpecificStats* specific = stage->getSpecificStats();
+    if (STAGE_COUNT_SCAN == stage->stageType()) {
+        const CountScanStats* spec = static_cast<const CountScanStats*>(specific);
+        const KeyPattern keyPattern{spec->keyPattern};
+        sb << " " << keyPattern;
+    } else if (STAGE_DISTINCT_SCAN == stage->stageType()) {
+        const DistinctScanStats* spec = static_cast<const DistinctScanStats*>(specific);
+        const KeyPattern keyPattern{spec->keyPattern};
+        sb << " " << keyPattern;
+    } else if (STAGE_GEO_NEAR_2D == stage->stageType()) {
+        const NearStats* spec = static_cast<const NearStats*>(specific);
+        const KeyPattern keyPattern{spec->keyPattern};
+        sb << " " << keyPattern;
+    } else if (STAGE_GEO_NEAR_2DSPHERE == stage->stageType()) {
+        const NearStats* spec = static_cast<const NearStats*>(specific);
+        const KeyPattern keyPattern{spec->keyPattern};
+        sb << " " << keyPattern;
+    } else if (STAGE_IXSCAN == stage->stageType()) {
+        const IndexScanStats* spec = static_cast<const IndexScanStats*>(specific);
+        const KeyPattern keyPattern{spec->keyPattern};
+        sb << " " << keyPattern;
+    } else if (STAGE_TEXT == stage->stageType()) {
+        const TextStats* spec = static_cast<const TextStats*>(specific);
+        const KeyPattern keyPattern{spec->indexPrefix};
+        sb << " " << keyPattern;
+    }
+}
+
+/**
+ * Traverses the tree rooted at 'root', and adds all nodes into the list 'flattened'. If a
+ * MultiPlanStage is encountered, only adds the best plan and its children to 'flattened'.
+ */
+void flattenExecTree(const PlanStage* root, std::vector<const PlanStage*>* flattened) {
+    flattened->push_back(root);
+
+    if (root->stageType() == STAGE_MULTI_PLAN) {
+        // Only add the winning plan from a MultiPlanStage.
+        auto mps = static_cast<const MultiPlanStage*>(root);
+        const PlanStage* winningStage = mps->getChildren()[mps->bestPlanIdx()].get();
+        return flattenExecTree(winningStage, flattened);
+    }
+
+    const auto& children = root->getChildren();
+    for (size_t i = 0; i < children.size(); ++i) {
+        flattenExecTree(children[i].get(), flattened);
+    }
+}
+
+/**
+ * Traverse the tree rooted at 'root', and add all tree nodes into the list 'flattened'.
+ */
+void flattenStatsTree(const PlanStageStats* root, std::vector<const PlanStageStats*>* flattened) {
+    invariant(root->stageType != STAGE_MULTI_PLAN);
+    flattened->push_back(root);
+    for (auto&& child : root->children) {
+        flattenStatsTree(child.get(), flattened);
+    }
+}
+
+/**
+ * Given the SpecificStats object for a stage and the type of the stage, returns the number of index
+ * keys examined by the stage.
+ */
+size_t getKeysExamined(StageType type, const SpecificStats* specific) {
+    if (STAGE_IXSCAN == type) {
+        const IndexScanStats* spec = static_cast<const IndexScanStats*>(specific);
+        return spec->keysExamined;
+    } else if (STAGE_IDHACK == type) {
+        const IDHackStats* spec = static_cast<const IDHackStats*>(specific);
+        return spec->keysExamined;
+    } else if (STAGE_COUNT_SCAN == type) {
+        const CountScanStats* spec = static_cast<const CountScanStats*>(specific);
+        return spec->keysExamined;
+    } else if (STAGE_DISTINCT_SCAN == type) {
+        const DistinctScanStats* spec = static_cast<const DistinctScanStats*>(specific);
+        return spec->keysExamined;
+    }
+
+    return 0;
+}
+
+/**
+ * Given the SpecificStats object for a stage and the type of the stage, returns the number of
+ * documents examined by the stage.
+ */
+size_t getDocsExamined(StageType type, const SpecificStats* specific) {
+    if (STAGE_COLLSCAN == type) {
+        const CollectionScanStats* spec = static_cast<const CollectionScanStats*>(specific);
+        return spec->docsTested;
+    } else if (STAGE_FETCH == type) {
+        const FetchStats* spec = static_cast<const FetchStats*>(specific);
+        return spec->docsExamined;
+    } else if (STAGE_IDHACK == type) {
+        const IDHackStats* spec = static_cast<const IDHackStats*>(specific);
+        return spec->docsExamined;
+    } else if (STAGE_TEXT_OR == type) {
+        const TextOrStats* spec = static_cast<const TextOrStats*>(specific);
+        return spec->fetches;
+    }
+
+    return 0;
+}
+
+/**
+ * Adds the path-level multikey information to the explain output in a field called "multiKeyPaths".
+ * The value associated with the "multiKeyPaths" field is an object with keys equal to those in the
+ * index key pattern and values equal to an array of strings corresponding to paths that cause the
+ * index to be multikey.
+ *
+ * For example, with the index {'a.b': 1, 'a.c': 1} where the paths "a" and "a.b" cause the
+ * index to be multikey, we'd have {'multiKeyPaths': {'a.b': ['a', 'a.b'], 'a.c': ['a']}}.
+ *
+ * This function should only be called if the associated index supports path-level multikey
+ * tracking.
+ */
+void appendMultikeyPaths(const BSONObj& keyPattern,
+                         const MultikeyPaths& multikeyPaths,
+                         BSONObjBuilder* bob) {
+    BSONObjBuilder subMultikeyPaths(bob->subobjStart("multiKeyPaths"));
+
+    size_t i = 0;
+    for (const auto keyElem : keyPattern) {
+        const FieldRef path{keyElem.fieldNameStringData()};
+
+        BSONArrayBuilder arrMultikeyComponents(
+            subMultikeyPaths.subarrayStart(keyElem.fieldNameStringData()));
+        for (const auto multikeyComponent : multikeyPaths[i]) {
+            arrMultikeyComponents.append(path.dottedSubstring(0, multikeyComponent + 1));
+        }
+        arrMultikeyComponents.doneFast();
+
+        ++i;
+    }
+
+    subMultikeyPaths.doneFast();
+}
+
+/**
+ * Converts the stats tree 'stats' into a corresponding BSON object containing explain information.
+ *
+ * Generates the BSON stats at a verbosity specified by 'verbosity'.
+ */
+void statsToBSON(const PlanStageStats& stats,
+                 ExplainOptions::Verbosity verbosity,
+                 BSONObjBuilder* bob,
+                 BSONObjBuilder* topLevelBob) {
+    invariant(bob);
+    invariant(topLevelBob);
+
+    // Stop as soon as the BSON object we're building exceeds 10 MB.
+    static const int kMaxStatsBSONSize = 10 * 1024 * 1024;
+    if (topLevelBob->len() > kMaxStatsBSONSize) {
+        bob->append("warning", "stats tree exceeded 10 MB");
+        return;
+    }
+
+    // Stage name.
+    bob->append("stage", stats.common.stageTypeStr);
+
+    // Display the BSON representation of the filter, if there is one.
+    if (!stats.common.filter.isEmpty()) {
+        bob->append("filter", stats.common.filter);
+    }
+
+    // Some top-level exec stats get pulled out of the root stage.
+    if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
+        bob->appendNumber("nReturned", stats.common.advanced);
+        // Include executionTimeMillis if it was recorded.
+        if (stats.common.executionTimeMillis) {
+            bob->appendNumber("executionTimeMillisEstimate", *stats.common.executionTimeMillis);
+        }
+
+        bob->appendNumber("works", stats.common.works);
+        bob->appendNumber("advanced", stats.common.advanced);
+        bob->appendNumber("needTime", stats.common.needTime);
+        bob->appendNumber("needYield", stats.common.needYield);
+        bob->appendNumber("saveState", stats.common.yields);
+        bob->appendNumber("restoreState", stats.common.unyields);
+        if (stats.common.failed)
+            bob->appendBool("failed", stats.common.failed);
+        bob->appendNumber("isEOF", stats.common.isEOF);
+    }
+
+    // Stage-specific stats
+    if (STAGE_AND_HASH == stats.stageType) {
+        AndHashStats* spec = static_cast<AndHashStats*>(stats.specific.get());
+
+        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
+            bob->appendNumber("memUsage", spec->memUsage);
+            bob->appendNumber("memLimit", spec->memLimit);
+
+            for (size_t i = 0; i < spec->mapAfterChild.size(); ++i) {
+                bob->appendNumber(std::string(str::stream() << "mapAfterChild_" << i),
+                                  spec->mapAfterChild[i]);
+            }
+        }
+    } else if (STAGE_AND_SORTED == stats.stageType) {
+        AndSortedStats* spec = static_cast<AndSortedStats*>(stats.specific.get());
+
+        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
+            for (size_t i = 0; i < spec->failedAnd.size(); ++i) {
+                bob->appendNumber(std::string(str::stream() << "failedAnd_" << i),
+                                  spec->failedAnd[i]);
+            }
+        }
+    } else if (STAGE_COLLSCAN == stats.stageType) {
+        CollectionScanStats* spec = static_cast<CollectionScanStats*>(stats.specific.get());
+        bob->append("direction", spec->direction > 0 ? "forward" : "backward");
+        if (spec->minTs) {
+            bob->append("minTs", *(spec->minTs));
+        }
+        if (spec->maxTs) {
+            bob->append("maxTs", *(spec->maxTs));
+        }
+        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
+            bob->appendNumber("docsExamined", spec->docsTested);
+        }
+    } else if (STAGE_COUNT == stats.stageType) {
+        CountStats* spec = static_cast<CountStats*>(stats.specific.get());
+
+        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
+            bob->appendNumber("nCounted", spec->nCounted);
+            bob->appendNumber("nSkipped", spec->nSkipped);
+        }
+    } else if (STAGE_COUNT_SCAN == stats.stageType) {
+        CountScanStats* spec = static_cast<CountScanStats*>(stats.specific.get());
+
+        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
+            bob->appendNumber("keysExamined", spec->keysExamined);
+        }
+
+        bob->append("keyPattern", spec->keyPattern);
+        bob->append("indexName", spec->indexName);
+        if (!spec->collation.isEmpty()) {
+            bob->append("collation", spec->collation);
+        }
+        bob->appendBool("isMultiKey", spec->isMultiKey);
+        if (!spec->multiKeyPaths.empty()) {
+            appendMultikeyPaths(spec->keyPattern, spec->multiKeyPaths, bob);
+        }
+        bob->appendBool("isUnique", spec->isUnique);
+        bob->appendBool("isSparse", spec->isSparse);
+        bob->appendBool("isPartial", spec->isPartial);
+        bob->append("indexVersion", spec->indexVersion);
+
+        BSONObjBuilder indexBoundsBob;
+        indexBoundsBob.append("startKey", spec->startKey);
+        indexBoundsBob.append("startKeyInclusive", spec->startKeyInclusive);
+        indexBoundsBob.append("endKey", spec->endKey);
+        indexBoundsBob.append("endKeyInclusive", spec->endKeyInclusive);
+        bob->append("indexBounds", indexBoundsBob.obj());
+    } else if (STAGE_DELETE == stats.stageType) {
+        DeleteStats* spec = static_cast<DeleteStats*>(stats.specific.get());
+
+        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
+            bob->appendNumber("nWouldDelete", spec->docsDeleted);
+        }
+    } else if (STAGE_DISTINCT_SCAN == stats.stageType) {
+        DistinctScanStats* spec = static_cast<DistinctScanStats*>(stats.specific.get());
+
+        bob->append("keyPattern", spec->keyPattern);
+        bob->append("indexName", spec->indexName);
+        if (!spec->collation.isEmpty()) {
+            bob->append("collation", spec->collation);
+        }
+        bob->appendBool("isMultiKey", spec->isMultiKey);
+        if (!spec->multiKeyPaths.empty()) {
+            appendMultikeyPaths(spec->keyPattern, spec->multiKeyPaths, bob);
+        }
+        bob->appendBool("isUnique", spec->isUnique);
+        bob->appendBool("isSparse", spec->isSparse);
+        bob->appendBool("isPartial", spec->isPartial);
+        bob->append("indexVersion", spec->indexVersion);
+        bob->append("direction", spec->direction > 0 ? "forward" : "backward");
+
+        if ((topLevelBob->len() + spec->indexBounds.objsize()) > kMaxStatsBSONSize) {
+            bob->append("warning", "index bounds omitted due to BSON size limit");
+        } else {
+            bob->append("indexBounds", spec->indexBounds);
+        }
+
+        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
+            bob->appendNumber("keysExamined", spec->keysExamined);
+        }
+    } else if (STAGE_ENSURE_SORTED == stats.stageType) {
+        EnsureSortedStats* spec = static_cast<EnsureSortedStats*>(stats.specific.get());
+
+        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
+            bob->appendNumber("nDropped", spec->nDropped);
+        }
+    } else if (STAGE_FETCH == stats.stageType) {
+        FetchStats* spec = static_cast<FetchStats*>(stats.specific.get());
+        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
+            bob->appendNumber("docsExamined", spec->docsExamined);
+            bob->appendNumber("alreadyHasObj", spec->alreadyHasObj);
+        }
+    } else if (STAGE_GEO_NEAR_2D == stats.stageType || STAGE_GEO_NEAR_2DSPHERE == stats.stageType) {
+        NearStats* spec = static_cast<NearStats*>(stats.specific.get());
+
+        bob->append("keyPattern", spec->keyPattern);
+        bob->append("indexName", spec->indexName);
+        bob->append("indexVersion", spec->indexVersion);
+
+        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
+            BSONArrayBuilder intervalsBob(bob->subarrayStart("searchIntervals"));
+            for (std::vector<IntervalStats>::const_iterator it = spec->intervalStats.begin();
+                 it != spec->intervalStats.end();
+                 ++it) {
+                BSONObjBuilder intervalBob(intervalsBob.subobjStart());
+                intervalBob.append("minDistance", it->minDistanceAllowed);
+                intervalBob.append("maxDistance", it->maxDistanceAllowed);
+                intervalBob.append("maxInclusive", it->inclusiveMaxDistanceAllowed);
+                intervalBob.appendNumber("nBuffered", it->numResultsBuffered);
+                intervalBob.appendNumber("nReturned", it->numResultsReturned);
+            }
+            intervalsBob.doneFast();
+        }
+    } else if (STAGE_IDHACK == stats.stageType) {
+        IDHackStats* spec = static_cast<IDHackStats*>(stats.specific.get());
+        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
+            bob->appendNumber("keysExamined", spec->keysExamined);
+            bob->appendNumber("docsExamined", spec->docsExamined);
+        }
+    } else if (STAGE_IXSCAN == stats.stageType) {
+        IndexScanStats* spec = static_cast<IndexScanStats*>(stats.specific.get());
+
+        bob->append("keyPattern", spec->keyPattern);
+        bob->append("indexName", spec->indexName);
+        if (!spec->collation.isEmpty()) {
+            bob->append("collation", spec->collation);
+        }
+        bob->appendBool("isMultiKey", spec->isMultiKey);
+        if (!spec->multiKeyPaths.empty()) {
+            appendMultikeyPaths(spec->keyPattern, spec->multiKeyPaths, bob);
+        }
+        bob->appendBool("isUnique", spec->isUnique);
+        bob->appendBool("isSparse", spec->isSparse);
+        bob->appendBool("isPartial", spec->isPartial);
+        bob->append("indexVersion", spec->indexVersion);
+        bob->append("direction", spec->direction > 0 ? "forward" : "backward");
+
+        if ((topLevelBob->len() + spec->indexBounds.objsize()) > kMaxStatsBSONSize) {
+            bob->append("warning", "index bounds omitted due to BSON size limit");
+        } else {
+            bob->append("indexBounds", spec->indexBounds);
+        }
+
+        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
+            bob->appendNumber("keysExamined", spec->keysExamined);
+            bob->appendNumber("seeks", spec->seeks);
+            bob->appendNumber("dupsTested", spec->dupsTested);
+            bob->appendNumber("dupsDropped", spec->dupsDropped);
+        }
+    } else if (STAGE_OR == stats.stageType) {
+        OrStats* spec = static_cast<OrStats*>(stats.specific.get());
+
+        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
+            bob->appendNumber("dupsTested", spec->dupsTested);
+            bob->appendNumber("dupsDropped", spec->dupsDropped);
+        }
+    } else if (STAGE_LIMIT == stats.stageType) {
+        LimitStats* spec = static_cast<LimitStats*>(stats.specific.get());
+        bob->appendNumber("limitAmount", spec->limit);
+    } else if (isProjectionStageType(stats.stageType)) {
+        ProjectionStats* spec = static_cast<ProjectionStats*>(stats.specific.get());
+        bob->append("transformBy", spec->projObj);
+    } else if (STAGE_RECORD_STORE_FAST_COUNT == stats.stageType) {
+        CountStats* spec = static_cast<CountStats*>(stats.specific.get());
+
+        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
+            bob->appendNumber("nCounted", spec->nCounted);
+            bob->appendNumber("nSkipped", spec->nSkipped);
+        }
+    } else if (STAGE_SHARDING_FILTER == stats.stageType) {
+        ShardingFilterStats* spec = static_cast<ShardingFilterStats*>(stats.specific.get());
+
+        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
+            bob->appendNumber("chunkSkips", spec->chunkSkips);
+        }
+    } else if (STAGE_SKIP == stats.stageType) {
+        SkipStats* spec = static_cast<SkipStats*>(stats.specific.get());
+        bob->appendNumber("skipAmount", spec->skip);
+    } else if (isSortStageType(stats.stageType)) {
+        SortStats* spec = static_cast<SortStats*>(stats.specific.get());
+        bob->append("sortPattern", spec->sortPattern);
+        bob->appendIntOrLL("memLimit", spec->maxMemoryUsageBytes);
+
+        if (spec->limit > 0) {
+            bob->appendIntOrLL("limitAmount", spec->limit);
+        }
+
+        bob->append("type", stats.stageType == STAGE_SORT_SIMPLE ? "simple" : "default");
+
+        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
+            bob->appendIntOrLL("totalDataSizeSorted", spec->totalDataSizeBytes);
+            bob->appendBool("usedDisk", spec->wasDiskUsed);
+        }
+    } else if (STAGE_SORT_MERGE == stats.stageType) {
+        MergeSortStats* spec = static_cast<MergeSortStats*>(stats.specific.get());
+        bob->append("sortPattern", spec->sortPattern);
+
+        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
+            bob->appendNumber("dupsTested", spec->dupsTested);
+            bob->appendNumber("dupsDropped", spec->dupsDropped);
+        }
+    } else if (STAGE_TEXT == stats.stageType) {
+        TextStats* spec = static_cast<TextStats*>(stats.specific.get());
+
+        bob->append("indexPrefix", spec->indexPrefix);
+        bob->append("indexName", spec->indexName);
+        bob->append("parsedTextQuery", spec->parsedTextQuery);
+        bob->append("textIndexVersion", spec->textIndexVersion);
+    } else if (STAGE_TEXT_MATCH == stats.stageType) {
+        TextMatchStats* spec = static_cast<TextMatchStats*>(stats.specific.get());
+
+        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
+            bob->appendNumber("docsRejected", spec->docsRejected);
+        }
+    } else if (STAGE_TEXT_OR == stats.stageType) {
+        TextOrStats* spec = static_cast<TextOrStats*>(stats.specific.get());
+
+        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
+            bob->appendNumber("docsExamined", spec->fetches);
+        }
+    } else if (STAGE_UPDATE == stats.stageType) {
+        UpdateStats* spec = static_cast<UpdateStats*>(stats.specific.get());
+
+        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
+            bob->appendNumber("nMatched", spec->nMatched);
+            bob->appendNumber("nWouldModify", spec->nModified);
+            bob->appendNumber("nWouldUpsert", spec->nUpserted);
+        }
+    }
+
+    // We're done if there are no children.
+    if (stats.children.empty()) {
+        return;
+    }
+
+    // If there's just one child (a common scenario), avoid making an array. This makes
+    // the output more readable by saving a level of nesting. Name the field 'inputStage'
+    // rather than 'inputStages'.
+    if (1 == stats.children.size()) {
+        BSONObjBuilder childBob;
+        statsToBSON(*stats.children[0], verbosity, &childBob, topLevelBob);
+        bob->append("inputStage", childBob.obj());
+        return;
+    }
+
+    // There is more than one child. Recursively call statsToBSON(...) on each
+    // of them and add them to the 'inputStages' array.
+
+    BSONArrayBuilder childrenBob(bob->subarrayStart("inputStages"));
+    for (size_t i = 0; i < stats.children.size(); ++i) {
+        BSONObjBuilder childBob(childrenBob.subobjStart());
+        statsToBSON(*stats.children[i], verbosity, &childBob, topLevelBob);
+    }
+    childrenBob.doneFast();
+}
+
+PlanSummaryStats collectExecutionStatsSummary(const PlanStageStats* stats) {
+    PlanSummaryStats summary;
+    summary.nReturned = stats->common.advanced;
+
+    if (stats->common.executionTimeMillis) {
+        summary.executionTimeMillisEstimate = *stats->common.executionTimeMillis;
+    }
+
+    // Flatten the stats tree into a list.
+    std::vector<const PlanStageStats*> statsNodes;
+    flattenStatsTree(stats, &statsNodes);
+
+    // Iterate over all stages in the tree and get the total number of keys/docs examined.
+    // These are just aggregations of information already available in the stats tree.
+    for (size_t i = 0; i < statsNodes.size(); ++i) {
+        summary.totalKeysExamined +=
+            getKeysExamined(statsNodes[i]->stageType, statsNodes[i]->specific.get());
+        summary.totalDocsExamined +=
+            getDocsExamined(statsNodes[i]->stageType, statsNodes[i]->specific.get());
+    }
+
+    summary.planFailed = stats->common.failed;
+
+    return summary;
+}
+}  // namespace
+
+bool PlanExplainerImpl::isMultiPlan() const {
+    return getStageByType(_root, StageType::STAGE_MULTI_PLAN) != nullptr;
+}
+
+std::string PlanExplainerImpl::getPlanSummary() const {
+    std::vector<const PlanStage*> stages;
+    flattenExecTree(_root, &stages);
+
+    // Use this stream to build the plan summary string.
+    StringBuilder sb;
+    bool seenLeaf = false;
+
+    for (size_t i = 0; i < stages.size(); i++) {
+        if (stages[i]->getChildren().empty()) {
+            // This is a leaf node. Add to the plan summary string accordingly. Unless
+            // this is the first leaf we've seen, add a delimiting string first.
+            if (seenLeaf) {
+                sb << ", ";
+            } else {
+                seenLeaf = true;
+            }
+            addStageSummaryStr(stages[i], sb);
+        }
+    }
+
+    return sb.str();
+}
+
+void PlanExplainerImpl::getSummaryStats(PlanSummaryStats* statsOut) const {
+    invariant(statsOut);
+
+    // We can get some of the fields we need from the common stats stored in the
+    // root stage of the plan tree.
+    const CommonStats* common = _root->getCommonStats();
+    statsOut->nReturned = common->advanced;
+
+    // The other fields are aggregations over the stages in the plan tree. We flatten
+    // the tree into a list and then compute these aggregations.
+    std::vector<const PlanStage*> stages;
+    flattenExecTree(_root, &stages);
+
+    statsOut->totalKeysExamined = 0;
+    statsOut->totalDocsExamined = 0;
+
+    for (size_t i = 0; i < stages.size(); i++) {
+        statsOut->totalKeysExamined +=
+            getKeysExamined(stages[i]->stageType(), stages[i]->getSpecificStats());
+        statsOut->totalDocsExamined +=
+            getDocsExamined(stages[i]->stageType(), stages[i]->getSpecificStats());
+
+        if (isSortStageType(stages[i]->stageType())) {
+            statsOut->hasSortStage = true;
+
+            auto sortStage = static_cast<const SortStage*>(stages[i]);
+            auto sortStats = static_cast<const SortStats*>(sortStage->getSpecificStats());
+            statsOut->usedDisk = sortStats->wasDiskUsed;
+        }
+
+        if (STAGE_IXSCAN == stages[i]->stageType()) {
+            const IndexScan* ixscan = static_cast<const IndexScan*>(stages[i]);
+            const IndexScanStats* ixscanStats =
+                static_cast<const IndexScanStats*>(ixscan->getSpecificStats());
+            statsOut->indexesUsed.insert(ixscanStats->indexName);
+        } else if (STAGE_COUNT_SCAN == stages[i]->stageType()) {
+            const CountScan* countScan = static_cast<const CountScan*>(stages[i]);
+            const CountScanStats* countScanStats =
+                static_cast<const CountScanStats*>(countScan->getSpecificStats());
+            statsOut->indexesUsed.insert(countScanStats->indexName);
+        } else if (STAGE_IDHACK == stages[i]->stageType()) {
+            const IDHackStage* idHackStage = static_cast<const IDHackStage*>(stages[i]);
+            const IDHackStats* idHackStats =
+                static_cast<const IDHackStats*>(idHackStage->getSpecificStats());
+            statsOut->indexesUsed.insert(idHackStats->indexName);
+        } else if (STAGE_DISTINCT_SCAN == stages[i]->stageType()) {
+            const DistinctScan* distinctScan = static_cast<const DistinctScan*>(stages[i]);
+            const DistinctScanStats* distinctScanStats =
+                static_cast<const DistinctScanStats*>(distinctScan->getSpecificStats());
+            statsOut->indexesUsed.insert(distinctScanStats->indexName);
+        } else if (STAGE_TEXT == stages[i]->stageType()) {
+            const TextStage* textStage = static_cast<const TextStage*>(stages[i]);
+            const TextStats* textStats =
+                static_cast<const TextStats*>(textStage->getSpecificStats());
+            statsOut->indexesUsed.insert(textStats->indexName);
+        } else if (STAGE_GEO_NEAR_2D == stages[i]->stageType() ||
+                   STAGE_GEO_NEAR_2DSPHERE == stages[i]->stageType()) {
+            const NearStage* nearStage = static_cast<const NearStage*>(stages[i]);
+            const NearStats* nearStats =
+                static_cast<const NearStats*>(nearStage->getSpecificStats());
+            statsOut->indexesUsed.insert(nearStats->indexName);
+        } else if (STAGE_CACHED_PLAN == stages[i]->stageType()) {
+            const CachedPlanStage* cachedPlan = static_cast<const CachedPlanStage*>(stages[i]);
+            const CachedPlanStats* cachedStats =
+                static_cast<const CachedPlanStats*>(cachedPlan->getSpecificStats());
+            statsOut->replanReason = cachedStats->replanReason;
+        } else if (STAGE_MULTI_PLAN == stages[i]->stageType()) {
+            statsOut->fromMultiPlanner = true;
+        } else if (STAGE_COLLSCAN == stages[i]->stageType()) {
+            statsOut->collectionScans++;
+            const auto collScan = static_cast<const CollectionScan*>(stages[i]);
+            const auto collScanStats =
+                static_cast<const CollectionScanStats*>(collScan->getSpecificStats());
+            if (!collScanStats->tailable)
+                statsOut->collectionScansNonTailable++;
+        }
+    }
+}
+
+PlanExplainer::PlanStatsDetails PlanExplainerImpl::getWinningPlanStats(
+    ExplainOptions::Verbosity verbosity) const {
+    auto stage = getStageByType(_root, StageType::STAGE_MULTI_PLAN);
+    invariant(stage == nullptr || stage->stageType() == StageType::STAGE_MULTI_PLAN);
+    auto mps = static_cast<MultiPlanStage*>(stage);
+
+    auto&& [stats, summary] =
+        [&]() -> std::pair<std::unique_ptr<PlanStageStats>, boost::optional<PlanSummaryStats>> {
+        auto stats =
+            mps ? std::move(mps->getStats()->children[mps->bestPlanIdx()]) : _root->getStats();
+
+        if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
+            return {std::move(stats), collectExecutionStatsSummary(stats.get())};
+        }
+
+        return {std::move(stats), boost::none};
+    }();
+
+    BSONObjBuilder bob;
+    statsToBSON(*stats, verbosity, &bob, &bob);
+    return {bob.obj(), std::move(summary)};
+}
+
+std::vector<PlanExplainer::PlanStatsDetails> PlanExplainerImpl::getRejectedPlansStats(
+    ExplainOptions::Verbosity verbosity) const {
+    auto stage = getStageByType(_root, StageType::STAGE_MULTI_PLAN);
+    invariant(stage == nullptr || stage->stageType() == StageType::STAGE_MULTI_PLAN);
+    auto mps = static_cast<MultiPlanStage*>(stage);
+
+    std::vector<PlanStatsDetails> res;
+
+    // Get the stats from the trial period for all the plans.
+    if (mps) {
+        const auto mpsStats = mps->getStats();
+        for (size_t i = 0; i < mpsStats->children.size(); ++i) {
+            if (i != static_cast<size_t>(mps->bestPlanIdx())) {
+                BSONObjBuilder bob;
+                statsToBSON(*mpsStats->children[i], verbosity, &bob, &bob);
+                res.push_back({bob.obj(),
+                               {verbosity >= ExplainOptions::Verbosity::kExecStats,
+                                collectExecutionStatsSummary(mpsStats->children[i].get())}});
+            }
+        }
+    }
+
+    return res;
+}
+
+std::vector<PlanExplainer::PlanStatsDetails> PlanExplainerImpl::getCachedPlanStats(
+    const PlanCacheEntry& entry, ExplainOptions::Verbosity verbosity) const {
+    std::vector<PlanStatsDetails> res;
+    for (auto&& stats : entry.decision->getStats<PlanStageStats>()) {
+        BSONObjBuilder bob;
+        statsToBSON(*stats, verbosity, &bob, &bob);
+        res.push_back({bob.obj(),
+                       {verbosity >= ExplainOptions::Verbosity::kExecStats,
+                        collectExecutionStatsSummary(stats.get())}});
+    }
+    return res;
+}
+
+PlanStage* getStageByType(PlanStage* root, StageType type) {
+    if (root->stageType() == type) {
+        return root;
+    }
+
+    const auto& children = root->getChildren();
+    for (size_t i = 0; i < children.size(); i++) {
+        PlanStage* result = getStageByType(children[i].get(), type);
+        if (result) {
+            return result;
+        }
+    }
+
+    return nullptr;
+}
+}  // namespace mongo
diff --git a/src/mongo/db/query/plan_explainer_impl.h b/src/mongo/db/query/plan_explainer_impl.h
new file mode 100644
index 00000000000..f70b584cc2d
--- /dev/null
+++ b/src/mongo/db/query/plan_explainer_impl.h
@@ -0,0 +1,65 @@
+/**
+ *    Copyright (C) 2020-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include "mongo/db/exec/plan_stage.h"
+#include "mongo/db/query/plan_explainer.h"
+
+namespace mongo {
+/**
+ * A PlanExplainer implementation for classic execution plans.
+ *
+ * For classic execution plans all information required to generate explain output in various
+ * formats is stored in the execution tree itself, so having access to the root stage of the
+ * execution tree this PlanExplainer should obtain all plan details and execution stats.
+ */
+class PlanExplainerImpl final : public PlanExplainer {
+public:
+    PlanExplainerImpl(PlanStage* root) : _root{root} {}
+
+    bool isMultiPlan() const final;
+    std::string getPlanSummary() const final;
+    void getSummaryStats(PlanSummaryStats* statsOut) const final;
+    PlanStatsDetails getWinningPlanStats(ExplainOptions::Verbosity verbosity) const final;
+    std::vector<PlanStatsDetails> getRejectedPlansStats(
+        ExplainOptions::Verbosity verbosity) const final;
+    std::vector<PlanStatsDetails> getCachedPlanStats(
+        const PlanCacheEntry& entry, ExplainOptions::Verbosity verbosity) const final;
+
+private:
+    PlanStage* const _root;
+};
+
+/**
+ * Retrieves the first stage of a given type from the plan tree, or nullptr if no such stage is
+ * found.
+ */
+PlanStage* getStageByType(PlanStage* root, StageType type);
+}  // namespace mongo
diff --git a/src/mongo/db/query/plan_explainer_sbe.cpp b/src/mongo/db/query/plan_explainer_sbe.cpp
new file mode 100644
index 00000000000..153c6b5814a
--- /dev/null
+++ b/src/mongo/db/query/plan_explainer_sbe.cpp
@@ -0,0 +1,61 @@
+/**
+ *    Copyright (C) 2020-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/db/query/plan_explainer_sbe.h"
+
+namespace mongo {
+std::string PlanExplainerSBE::getPlanSummary() const {
+    // TODO: SERVER-50743
+    return "unsupported";
+}
+
+void PlanExplainerSBE::getSummaryStats(PlanSummaryStats* statsOut) const {
+    // TODO: SERVER-50744
+}
+
+PlanExplainer::PlanStatsDetails PlanExplainerSBE::getWinningPlanStats(
+    ExplainOptions::Verbosity verbosity) const {
+    // TODO: SERVER-50728
+    return {{}, {}};
+}
+
+std::vector<PlanExplainer::PlanStatsDetails> PlanExplainerSBE::getRejectedPlansStats(
+    ExplainOptions::Verbosity verbosity) const {
+    // TODO: SERVER-50728
+    return {};
+}
+
+std::vector<PlanExplainer::PlanStatsDetails> PlanExplainerSBE::getCachedPlanStats(
+    const PlanCacheEntry& entry, ExplainOptions::Verbosity verbosity) const {
+    // TODO: SERVER-50728
+    return {};
+}
+}  // namespace mongo
diff --git a/src/mongo/db/query/plan_explainer_sbe.h b/src/mongo/db/query/plan_explainer_sbe.h
new file mode 100644
index 00000000000..519d15df13e
--- /dev/null
+++ b/src/mongo/db/query/plan_explainer_sbe.h
@@ -0,0 +1,56 @@
+/**
+ *    Copyright (C) 2020-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include "mongo/db/exec/sbe/stages/stages.h"
+#include "mongo/db/query/plan_explainer.h"
+#include "mongo/db/query/query_solution.h"
+
+namespace mongo {
+/**
+ * A PlanExplainer implementation for SBE execution plans.
+ */
+class PlanExplainerSBE final : public PlanExplainer {
+public:
+    PlanExplainerSBE(const sbe::PlanStage* root, const QuerySolution* solution) {}
+
+    bool isMultiPlan() const final {
+        return false;
+    }
+
+    std::string getPlanSummary() const final;
+    void getSummaryStats(PlanSummaryStats* statsOut) const final;
+    PlanStatsDetails getWinningPlanStats(ExplainOptions::Verbosity verbosity) const final;
+    std::vector<PlanStatsDetails> getRejectedPlansStats(
+        ExplainOptions::Verbosity verbosity) const final;
+    std::vector<PlanStatsDetails> getCachedPlanStats(
+        const PlanCacheEntry& entry, ExplainOptions::Verbosity verbosity) const final;
+};
+}  // namespace mongo
diff --git a/src/mongo/db/query/plan_ranker.h b/src/mongo/db/query/plan_ranker.h
index d9df7c8557a..4fc39a1e852 100644
--- a/src/mongo/db/query/plan_ranker.h
+++ b/src/mongo/db/query/plan_ranker.h
@@ -35,6 +35,7 @@
 #include "mongo/db/exec/sbe/stages/plan_stats.h"
 #include "mongo/db/exec/working_set.h"
 #include "mongo/db/query/explain.h"
+#include "mongo/db/query/plan_explainer_factory.h"
 #include "mongo/db/query/query_solution.h"
 #include "mongo/util/container_size_helper.h"
 
@@ -327,10 +328,16 @@ StatusWith<std::unique_ptr<PlanRankingDecision>> pickBestPlan(
         if (!candidates[i].failed) {
             log_detail::logScoringPlan(
                 [& candidate = candidates[i]]() { return candidate.solution->toString(); },
-                [& stats = *statTrees[i]]() {
-                    return Explain::statsToBSON(stats).jsonString(ExtendedRelaxedV2_0_0, true);
+                [root = &*candidates[i].root, solution = candidates[i].solution.get()]() {
+                    auto explainer = plan_explainer_factory::makePlanExplainer(root, solution);
+                    auto&& [stats, _] =
+                        explainer->getWinningPlanStats(ExplainOptions::Verbosity::kExecStats);
+                    return stats.jsonString(ExtendedRelaxedV2_0_0, true);
+                },
+                [root = &*candidates[i].root, solution = candidates[i].solution.get()]() {
+                    auto explainer = plan_explainer_factory::makePlanExplainer(root, solution);
+                    return explainer->getPlanSummary();
                 },
-                [root = &*candidates[i].root]() { return Explain::getPlanSummary(root); },
                 i,
                 statTrees[i]->common.isEOF);
             auto scorer = [solution = candidates[i].solution.get()]()
@@ -353,7 +360,10 @@ StatusWith<std::unique_ptr<PlanRankingDecision>> pickBestPlan(
         } else {
             failed.push_back(i);
             log_detail::logFailedPlan(
-                [root = &*candidates[i].root] { return Explain::getPlanSummary(root); });
+                [root = &*candidates[i].root, solution = candidates[i].solution.get()] {
+                    auto explainer = plan_explainer_factory::makePlanExplainer(root, solution);
+                    return explainer->getPlanSummary();
+                });
         }
     }
 
diff --git a/src/mongo/db/query/plan_summary_stats.h b/src/mongo/db/query/plan_summary_stats.h
index 4660c097795..8d0603baee7 100644
--- a/src/mongo/db/query/plan_summary_stats.h
+++ b/src/mongo/db/query/plan_summary_stats.h
@@ -55,12 +55,18 @@ struct PlanSummaryStats {
     // more than one collection scan may happen during execution (e.g. for $lookup execution).
     long long collectionScansNonTailable = 0;
 
+    // Time elapsed while executing this plan.
+    long long executionTimeMillisEstimate = 0;
+
     // Did this plan use an in-memory sort stage?
     bool hasSortStage = false;
 
     // Did this plan use disk space?
     bool usedDisk = false;
 
+    // Did this plan failed during execution?
+    bool planFailed = false;
+
     // The names of each index used by the plan.
     std::set<std::string> indexesUsed;
 
diff --git a/src/mongo/db/query/sbe_cached_solution_planner.cpp b/src/mongo/db/query/sbe_cached_solution_planner.cpp
index 17980a75de2..e8f4c263c1b 100644
--- a/src/mongo/db/query/sbe_cached_solution_planner.cpp
+++ b/src/mongo/db/query/sbe_cached_solution_planner.cpp
@@ -55,11 +55,13 @@ plan_ranker::CandidatePlan CachedSolutionPlanner::plan(
     if (candidate.failed) {
         // On failure, fall back to replanning the whole query. We neither evict the existing cache
         // entry, nor cache the result of replanning.
+        auto explainer = plan_explainer_factory::makePlanExplainer(candidate.root.get(),
+                                                                   candidate.solution.get());
         LOGV2_DEBUG(2057901,
                     1,
                     "Execution of cached plan failed, falling back to replan",
                     "query"_attr = redact(_cq.toStringShort()),
-                    "planSummary"_attr = Explain::getPlanSummary(candidate.root.get()));
+                    "planSummary"_attr = explainer->getPlanSummary());
         return replan(false);
     }
 
@@ -73,6 +75,8 @@ plan_ranker::CandidatePlan CachedSolutionPlanner::plan(
 
     // If we're here, the trial period took more than 'maxReadsBeforeReplan' physical reads. This
     // plan may not be efficient any longer, so we replan from scratch.
+    auto explainer =
+        plan_explainer_factory::makePlanExplainer(candidate.root.get(), candidate.solution.get());
     LOGV2_DEBUG(
         2058001,
         1,
@@ -81,7 +85,7 @@ plan_ranker::CandidatePlan CachedSolutionPlanner::plan(
         "maxReadsBeforeReplan"_attr = numReads,
         "decisionReads"_attr = _decisionReads,
         "query"_attr = redact(_cq.toStringShort()),
-        "planSummary"_attr = Explain::getPlanSummary(candidate.root.get()));
+        "planSummary"_attr = explainer->getPlanSummary());
     return replan(true);
 }
 
@@ -118,12 +122,14 @@ plan_ranker::CandidatePlan CachedSolutionPlanner::replan(bool shouldCache) const
         auto&& [root, data] = stage_builder::buildSlotBasedExecutableTree(
             _opCtx, _collection, _cq, *solutions[0], _yieldPolicy, true);
         prepareExecutionPlan(root.get(), &data);
+
+        auto explainer = plan_explainer_factory::makePlanExplainer(root.get(), solutions[0].get());
         LOGV2_DEBUG(
             2058101,
             1,
             "Replanning of query resulted in a single query solution, which will not be cached. ",
             "query"_attr = redact(_cq.toStringShort()),
-            "planSummary"_attr = Explain::getPlanSummary(root.get()),
+            "planSummary"_attr = explainer->getPlanSummary(),
             "shouldCache"_attr = (shouldCache ? "yes" : "no"));
         return {std::move(solutions[0]), std::move(root), std::move(data)};
     }
@@ -144,11 +150,13 @@ plan_ranker::CandidatePlan CachedSolutionPlanner::replan(bool shouldCache) const
         shouldCache ? PlanCachingMode::AlwaysCache : PlanCachingMode::NeverCache;
     MultiPlanner multiPlanner{_opCtx, _collection, _cq, cachingMode, _yieldPolicy};
     auto plan = multiPlanner.plan(std::move(solutions), std::move(roots));
+    auto explainer =
+        plan_explainer_factory::makePlanExplainer(plan.root.get(), plan.solution.get());
     LOGV2_DEBUG(2058201,
                 1,
                 "Query plan after replanning and its cache status",
                 "query"_attr = redact(_cq.toStringShort()),
-                "planSummary"_attr = Explain::getPlanSummary(plan.root.get()),
+                "planSummary"_attr = explainer->getPlanSummary(),
                 "shouldCache"_attr = (shouldCache ? "yes" : "no"));
     return plan;
 }
diff --git a/src/mongo/db/query/sbe_multi_planner.cpp b/src/mongo/db/query/sbe_multi_planner.cpp
index e07cc01f2e7..90f2ab8370c 100644
--- a/src/mongo/db/query/sbe_multi_planner.cpp
+++ b/src/mongo/db/query/sbe_multi_planner.cpp
@@ -63,10 +63,10 @@ plan_ranker::CandidatePlan MultiPlanner::finalizeExecutionPlans(
 
     LOGV2_DEBUG(
         4822875, 5, "Winning solution", "bestSolution"_attr = redact(winner.solution->toString()));
-    LOGV2_DEBUG(4822876,
-                2,
-                "Winning plan",
-                "planSummary"_attr = Explain::getPlanSummary(winner.root.get()));
+
+    auto explainer =
+        plan_explainer_factory::makePlanExplainer(winner.root.get(), winner.solution.get());
+    LOGV2_DEBUG(4822876, 2, "Winning plan", "planSummary"_attr = explainer->getPlanSummary());
 
     // Close all candidate plans but the winner.
     for (size_t ix = 1; ix < decision->candidateOrder.size(); ++ix) {
diff --git a/src/mongo/db/s/range_deletion_util.cpp b/src/mongo/db/s/range_deletion_util.cpp
index ec61ce0d47c..2a7f7cf5891 100644
--- a/src/mongo/db/s/range_deletion_util.cpp
+++ b/src/mongo/db/s/range_deletion_util.cpp
@@ -212,6 +212,9 @@ StatusWith<int> deleteNextBatch(OperationContext* opCtx,
         try {
             state = exec->getNext(&deletedObj, nullptr);
         } catch (const DBException& ex) {
+            auto&& explainer = exec->getPlanExplainer();
+            auto&& [stats, _] =
+                explainer.getWinningPlanStats(ExplainOptions::Verbosity::kExecStats);
             LOGV2_WARNING(23776,
                           "Cursor error while trying to delete {min} to {max} in {namespace}, "
                           "stats: {stats}, error: {error}",
@@ -219,7 +222,7 @@ StatusWith<int> deleteNextBatch(OperationContext* opCtx,
                           "min"_attr = redact(min),
                           "max"_attr = redact(max),
                           "namespace"_attr = nss,
-                          "stats"_attr = redact(exec->getStats()),
+                          "stats"_attr = redact(stats),
                           "error"_attr = redact(ex.toStatus()));
             throw;
         }