SERVER-62676 Added execution stats for `$lookup` pushed down into SBE

14 files changed, 630 insertions, 32 deletions

diff --git a/src/mongo/db/exec/plan_stats_visitor.h b/src/mongo/db/exec/plan_stats_visitor.h
index ade88fd2a5f..976f989b976 100644
--- a/src/mongo/db/exec/plan_stats_visitor.h
+++ b/src/mongo/db/exec/plan_stats_visitor.h
@@ -44,6 +44,7 @@ struct CheckBoundsStats;
 struct LoopJoinStats;
 struct TraverseStats;
 struct HashAggStats;
+struct HashLookupStats;
 }  // namespace sbe
 
 struct AndHashStats;
@@ -104,6 +105,7 @@ public:
     virtual void visit(tree_walker::MaybeConstPtr<IsConst, sbe::LoopJoinStats> stats) = 0;
     virtual void visit(tree_walker::MaybeConstPtr<IsConst, sbe::TraverseStats> stats) = 0;
     virtual void visit(tree_walker::MaybeConstPtr<IsConst, sbe::HashAggStats> stats) = 0;
+    virtual void visit(tree_walker::MaybeConstPtr<IsConst, sbe::HashLookupStats> stats) = 0;
 
     virtual void visit(tree_walker::MaybeConstPtr<IsConst, AndHashStats> stats) = 0;
     virtual void visit(tree_walker::MaybeConstPtr<IsConst, AndSortedStats> stats) = 0;
@@ -157,6 +159,7 @@ struct PlanStatsVisitorBase : public PlanStatsVisitor<IsConst> {
     void visit(tree_walker::MaybeConstPtr<IsConst, sbe::LoopJoinStats> stats) override {}
     void visit(tree_walker::MaybeConstPtr<IsConst, sbe::TraverseStats> stats) override {}
     void visit(tree_walker::MaybeConstPtr<IsConst, sbe::HashAggStats> stats) override {}
+    void visit(tree_walker::MaybeConstPtr<IsConst, sbe::HashLookupStats> stats) override {}
 
     void visit(tree_walker::MaybeConstPtr<IsConst, AndHashStats> stats) override {}
     void visit(tree_walker::MaybeConstPtr<IsConst, AndSortedStats> stats) override {}
diff --git a/src/mongo/db/exec/sbe/sbe_hash_lookup_test.cpp b/src/mongo/db/exec/sbe/sbe_hash_lookup_test.cpp
index 8df296d7729..eecd8037246 100644
--- a/src/mongo/db/exec/sbe/sbe_hash_lookup_test.cpp
+++ b/src/mongo/db/exec/sbe/sbe_hash_lookup_test.cpp
@@ -109,7 +109,6 @@ public:
                                                   collatorSlot,
                                                   kEmptyPlanNodeId);
 
-        stream << "-- OUTPUT ";
         StageResultsPrinters::SlotNames slotNames;
         if (outerKeyOnly) {
             slotNames.emplace_back(outerScanSlots[1], "outer_key");
@@ -152,6 +151,8 @@ public:
         StageResultsPrinters::make(firstStream, printOptions)
             .printStageResults(ctx, slotNames, stage);
         std::string firstStr = firstStream.str();
+        stream << "--- First Stats" << std::endl;
+        printHashLookupStats(stream, stage);
 
         // Execute the stage after reopen and verify that output is the same.
         stage->open(true);
@@ -160,6 +161,8 @@ public:
             .printStageResults(ctx, slotNames, stage);
         std::string secondStr = secondStream.str();
         ASSERT_EQ(firstStr, secondStr);
+        stream << "--- Second Stats" << std::endl;
+        printHashLookupStats(stream, stage);
 
         // Execute the stage after close and open and verify that output is the same.
         stage->close();
@@ -169,11 +172,12 @@ public:
             .printStageResults(ctx, slotNames, stage);
         std::string thirdStr = thirdStream.str();
         ASSERT_EQ(firstStr, thirdStr);
+        stream << "--- Third Stats" << std::endl;
+        printHashLookupStats(stream, stage);
 
-        // Execute the stage with spilling to disk.
         stage->close();
-        stage->open(false);
 
+        // Execute the stage with spilling to disk.
         auto defaultInternalQuerySBELookupApproxMemoryUseInBytesBeforeSpill =
             internalQuerySBELookupApproxMemoryUseInBytesBeforeSpill.load();
         internalQuerySBELookupApproxMemoryUseInBytesBeforeSpill.store(10);
@@ -181,11 +185,27 @@ public:
             internalQuerySBELookupApproxMemoryUseInBytesBeforeSpill.store(
                 defaultInternalQuerySBELookupApproxMemoryUseInBytesBeforeSpill);
         });
+
+        // TODO SERVER-65420: to enable the spilling tests, you need to uncomment the three lines
+        // below to clone the stage and prepare it again, as
+        // 'internalQuerySBELookupApproxMemoryUseInBytesBeforeSpill' is only considered at
+        // construction time of a `HashLookup` stage.
+        // auto stageUnqPtr = stage->clone();
+        // stage = stageUnqPtr.get();
+        // prepareTree(ctx, stage);
+
+        // TODO SERVER-65420: after enabling the spilling tests, there's no need to open the stage
+        // again, as it already get opened in the prepareTree call (above). Hence, you can remove
+        // the line below. stage->open(false);
+        stage->open(false);
+
         std::stringstream fourthStream;
         StageResultsPrinters::make(fourthStream, printOptions)
             .printStageResults(ctx, slotNames, stage);
         std::string fourthStr = fourthStream.str();
         ASSERT_EQ(firstStr, fourthStr);
+        stream << "--- Fourth Stats" << std::endl;
+        printHashLookupStats(stream, stage);
 
         // Execute the stage after reopen and we have spilled to disk and verify that output is the
         // same.
@@ -195,14 +215,32 @@ public:
             .printStageResults(ctx, slotNames, stage);
         std::string fifthStr = fifthStream.str();
         ASSERT_EQ(firstStr, fifthStr);
+        stream << "--- Fifth Stats" << std::endl;
+        printHashLookupStats(stream, stage);
+        stream << std::endl;
 
         stage->close();
+
+        stream << "-- OUTPUT ";
         stream << firstStr;
     }
 
 public:
     PrintOptions printOptions =
         PrintOptions().arrayObjectOrNestingMaxDepth(SIZE_MAX).useTagForAmbiguousValues(true);
+
+private:
+    static void printHashLookupStats(std::ostream& stream, const PlanStage* stage) {
+        auto stats = static_cast<const HashLookupStats*>(stage->getSpecificStats());
+        printHashLookupStats(stream, stats);
+    }
+    static void printHashLookupStats(std::ostream& stream, const HashLookupStats* stats) {
+        stream << "dsk:" << stats->usedDisk << std::endl;
+        stream << "htRecs:" << stats->spilledHtRecords << std::endl;
+        stream << "htIndices:" << stats->spilledHtBytesOverAllRecords << std::endl;
+        stream << "buffRecs:" << stats->spilledBuffRecords << std::endl;
+        stream << "buffBytes:" << stats->spilledBuffBytesOverAllRecords << std::endl;
+    }
 };
 
 TEST_F(HashLookupStageTest, BasicTests) {
diff --git a/src/mongo/db/exec/sbe/stages/hash_lookup.cpp b/src/mongo/db/exec/sbe/stages/hash_lookup.cpp
index b9262455def..73bfd564f71 100644
--- a/src/mongo/db/exec/sbe/stages/hash_lookup.cpp
+++ b/src/mongo/db/exec/sbe/stages/hash_lookup.cpp
@@ -30,6 +30,7 @@
 #include "mongo/platform/basic.h"
 
 #include "mongo/db/exec/sbe/stages/hash_lookup.h"
+#include "mongo/db/exec/sbe/stages/stage_visitors.h"
 
 #include "mongo/db/exec/sbe/expressions/expression.h"
 #include "mongo/db/exec/sbe/size_estimator.h"
@@ -249,7 +250,7 @@ void HashLookupStage::addHashTableEntry(value::SlotAccessor* keyAccessor, size_t
             _computedTotalMemUsage = newMemUsage;
         } else {
             // Write record to rs.
-            if (!_recordStoreHt) {
+            if (!hasSpilledHtToDisk()) {
                 makeTemporaryRecordStore();
             }
 
@@ -298,6 +299,8 @@ void HashLookupStage::makeTemporaryRecordStore() {
 
     _recordStoreHt = _opCtx->getServiceContext()->getStorageEngine()->makeTemporaryRecordStore(
         _opCtx, KeyFormat::String);
+
+    _specificStats.usedDisk = true;
 }
 
 void HashLookupStage::spillBufferedValueToDisk(OperationContext* opCtx,
@@ -318,6 +321,10 @@ void HashLookupStage::spillBufferedValueToDisk(OperationContext* opCtx,
     tassert(6373906,
             str::stream() << "Failed to write to disk because " << status.getStatus().reason(),
             status.isOK());
+
+    _specificStats.spilledBuffRecords++;
+    // Add size of record ID + size of buffer.
+    _specificStats.spilledBuffBytesOverAllRecords += sizeof(size_t) + buf.len();
     return;
 }
 
@@ -328,7 +335,7 @@ size_t HashLookupStage::bufferValueOrSpill(value::MaterializedRow& value) {
         _buffer.emplace_back(std::move(value));
         _computedTotalMemUsage = newMemUsage;
     } else {
-        if (!_recordStoreHt) {
+        if (!hasSpilledBufToDisk()) {
             makeTemporaryRecordStore();
         }
         spillBufferedValueToDisk(_opCtx, _recordStoreBuf->rs(), bufferIndex, value);
@@ -449,6 +456,15 @@ void HashLookupStage::writeIndicesToRecordStore(RecordStore* rs,
     auto [rid, typeBits] = serializeKeyForRecordStore(key);
 
     upsertToRecordStore(_opCtx, rs, rid, buf, typeBits, update);
+    if (!update) {
+        _specificStats.spilledHtRecords++;
+        // Add the size of key (which comprises of the memory usage for the key + its type bits),
+        // as well as the size of one integer to store the length of indices vector in the value.
+        _specificStats.spilledHtBytesOverAllRecords +=
+            rid.memUsage() + typeBits.getSize() + sizeof(size_t);
+    }
+    // Add the size of indices vector used in the hash-table value to the accounting.
+    _specificStats.spilledHtBytesOverAllRecords += value.size() * sizeof(size_t);
 }
 
 boost::optional<std::vector<size_t>> HashLookupStage::readIndicesFromRecordStore(
@@ -481,8 +497,11 @@ void HashLookupStage::spillIndicesToRecordStore(RecordStore* rs,
 
     auto update = false;
     if (valFromRs) {
-        update = true;
         valFromRs->insert(valFromRs->end(), value.begin(), value.end());
+        update = true;
+        // As we're updating these records, we'd remove the old size from the accounting. The new
+        // size is added back to the accounting in the call to 'writeIndicesToRecordStore' below.
+        _specificStats.spilledHtBytesOverAllRecords -= value.size();
     } else {
         valFromRs = value;
     }
@@ -570,11 +589,20 @@ std::unique_ptr<PlanStageStats> HashLookupStage::getStats(bool includeDebugInfo)
     auto ret = std::make_unique<PlanStageStats>(_commonStats);
     ret->children.emplace_back(outerChild()->getStats(includeDebugInfo));
     ret->children.emplace_back(innerChild()->getStats(includeDebugInfo));
+    ret->specific = std::make_unique<HashLookupStats>(_specificStats);
+    if (includeDebugInfo) {
+        BSONObjBuilder bob(StorageAccessStatsVisitor::collectStats(*this, ret.get()).toBSON());
+        // Spilling stats.
+        bob.appendBool("usedDisk", _specificStats.usedDisk)
+            .appendNumber("spilledRecords", _specificStats.getSpilledRecords())
+            .appendNumber("spilledBytesApprox", _specificStats.getSpilledBytesApprox());
+        ret->debugInfo = bob.obj();
+    }
     return ret;
 }
 
 const SpecificStats* HashLookupStage::getSpecificStats() const {
-    return nullptr;
+    return &_specificStats;
 }
 
 std::vector<DebugPrinter::Block> HashLookupStage::debugPrint() const {
diff --git a/src/mongo/db/exec/sbe/stages/hash_lookup.h b/src/mongo/db/exec/sbe/stages/hash_lookup.h
index e1694766500..5781e63af5a 100644
--- a/src/mongo/db/exec/sbe/stages/hash_lookup.h
+++ b/src/mongo/db/exec/sbe/stages/hash_lookup.h
@@ -231,5 +231,7 @@ private:
 
     std::unique_ptr<TemporaryRecordStore> _recordStoreHt;
     std::unique_ptr<TemporaryRecordStore> _recordStoreBuf;
+
+    HashLookupStats _specificStats;
 };
 }  // namespace mongo::sbe
diff --git a/src/mongo/db/exec/sbe/stages/ix_scan.h b/src/mongo/db/exec/sbe/stages/ix_scan.h
index 085cc04e81e..7e6c84ba14a 100644
--- a/src/mongo/db/exec/sbe/stages/ix_scan.h
+++ b/src/mongo/db/exec/sbe/stages/ix_scan.h
@@ -97,6 +97,11 @@ public:
     const SpecificStats* getSpecificStats() const final;
     std::vector<DebugPrinter::Block> debugPrint() const final;
     size_t estimateCompileTimeSize() const final;
+    std::string getIndexName() const {
+        // Note: the implementation of this getter cannot be moved to 'ix_scan.cpp', as it'd create
+        // a circular dependency after fixing the compilation scripts.
+        return _indexName;
+    }
 
 protected:
     void doSaveState(bool relinquishCursor) override;
diff --git a/src/mongo/db/exec/sbe/stages/loop_join.cpp b/src/mongo/db/exec/sbe/stages/loop_join.cpp
index 9f26e0b84d0..78607a246f9 100644
--- a/src/mongo/db/exec/sbe/stages/loop_join.cpp
+++ b/src/mongo/db/exec/sbe/stages/loop_join.cpp
@@ -30,6 +30,7 @@
 #include "mongo/platform/basic.h"
 
 #include "mongo/db/exec/sbe/stages/loop_join.h"
+#include "mongo/db/exec/sbe/stages/stage_visitors.h"
 
 #include "mongo/db/exec/sbe/size_estimator.h"
 #include "mongo/util/str.h"
@@ -173,22 +174,22 @@ void LoopJoinStage::doSaveState(bool relinquishCursor) {
 
 std::unique_ptr<PlanStageStats> LoopJoinStage::getStats(bool includeDebugInfo) const {
     auto ret = std::make_unique<PlanStageStats>(_commonStats);
+    ret->children.emplace_back(_children[0]->getStats(includeDebugInfo));
+    ret->children.emplace_back(_children[1]->getStats(includeDebugInfo));
+    ret->specific = std::make_unique<LoopJoinStats>(_specificStats);
 
     if (includeDebugInfo) {
-        BSONObjBuilder bob;
-        bob.appendNumber("innerOpens", static_cast<long long>(_specificStats.innerOpens));
-        bob.appendNumber("innerCloses", static_cast<long long>(_specificStats.innerCloses));
-        bob.append("outerProjects", _outerProjects.begin(), _outerProjects.end());
-        bob.append("outerCorrelated", _outerCorrelated.begin(), _outerCorrelated.end());
+        BSONObjBuilder bob(StorageAccessStatsVisitor::collectStats(*this, ret.get()).toBSON());
+        bob.appendNumber("innerOpens", static_cast<long long>(_specificStats.innerOpens))
+            .appendNumber("innerCloses", static_cast<long long>(_specificStats.innerCloses))
+            .append("outerProjects", _outerProjects.begin(), _outerProjects.end())
+            .append("outerCorrelated", _outerCorrelated.begin(), _outerCorrelated.end());
         if (_predicate) {
             bob.append("predicate", DebugPrinter{}.print(_predicate->debugPrint()));
         }
 
         ret->debugInfo = bob.obj();
     }
-
-    ret->children.emplace_back(_children[0]->getStats(includeDebugInfo));
-    ret->children.emplace_back(_children[1]->getStats(includeDebugInfo));
     return ret;
 }
 
diff --git a/src/mongo/db/exec/sbe/stages/plan_stats.cpp b/src/mongo/db/exec/sbe/stages/plan_stats.cpp
index eb35036d3e2..71a22275664 100644
--- a/src/mongo/db/exec/sbe/stages/plan_stats.cpp
+++ b/src/mongo/db/exec/sbe/stages/plan_stats.cpp
@@ -35,6 +35,7 @@
 
 #include "mongo/db/exec/plan_stats_walker.h"
 #include "mongo/db/exec/sbe/stages/plan_stats.h"
+#include "mongo/db/query/plan_summary_stats_visitor.h"
 #include "mongo/db/query/tree_walker.h"
 
 namespace mongo::sbe {
@@ -44,4 +45,20 @@ size_t calculateNumberOfReads(const PlanStageStats* root) {
     tree_walker::walk<true, PlanStageStats>(root, &walker);
     return visitor.numReads;
 }
+
+PlanSummaryStats collectExecutionStatsSummary(const PlanStageStats* root) {
+    invariant(root);
+
+    PlanSummaryStats summary;
+    summary.nReturned = root->common.advances;
+
+    if (root->common.executionTimeMillis) {
+        summary.executionTimeMillisEstimate = *root->common.executionTimeMillis;
+    }
+
+    auto visitor = PlanSummaryStatsVisitor(summary);
+    auto walker = PlanStageStatsWalker<true, CommonStats>(nullptr, nullptr, &visitor);
+    tree_walker::walk<true, PlanStageStats>(root, &walker);
+    return summary;
+}
 }  // namespace mongo::sbe
diff --git a/src/mongo/db/exec/sbe/stages/plan_stats.h b/src/mongo/db/exec/sbe/stages/plan_stats.h
index 46d52726b6c..263a4f86660 100644
--- a/src/mongo/db/exec/sbe/stages/plan_stats.h
+++ b/src/mongo/db/exec/sbe/stages/plan_stats.h
@@ -282,6 +282,38 @@ struct HashAggStats : public SpecificStats {
     long long lastSpilledRecordSize{0};
 };
 
+struct HashLookupStats : public SpecificStats {
+    std::unique_ptr<SpecificStats> clone() const final {
+        return std::make_unique<HashLookupStats>(*this);
+    }
+
+    uint64_t estimateObjectSizeInBytes() const final {
+        return sizeof(*this);
+    }
+
+    void acceptVisitor(PlanStatsConstVisitor* visitor) const final {
+        visitor->visit(this);
+    }
+
+    void acceptVisitor(PlanStatsMutableVisitor* visitor) final {
+        visitor->visit(this);
+    }
+
+    long long getSpilledRecords() const {
+        return spilledHtRecords + spilledBuffRecords;
+    }
+
+    long long getSpilledBytesApprox() const {
+        return spilledHtBytesOverAllRecords + spilledBuffBytesOverAllRecords;
+    }
+
+    bool usedDisk{false};
+    long long spilledHtRecords{0};
+    long long spilledHtBytesOverAllRecords{0};
+    long long spilledBuffRecords{0};
+    long long spilledBuffBytesOverAllRecords{0};
+};
+
 /**
  * Visitor for calculating the number of storage reads during plan execution.
  */
@@ -304,4 +336,9 @@ struct PlanStatsNumReadsVisitor : PlanStatsVisitorBase<true> {
  * a physical read (e.g. COLLSCAN or IXSCAN), then its 'numReads' stats is added to the total.
  */
 size_t calculateNumberOfReads(const PlanStageStats* root);
+
+/**
+ * Accumulates the summary of all execution statistics by walking over the specific-stats of stages.
+ */
+PlanSummaryStats collectExecutionStatsSummary(const PlanStageStats* root);
 }  // namespace mongo::sbe
diff --git a/src/mongo/db/exec/sbe/stages/stage_visitors.h b/src/mongo/db/exec/sbe/stages/stage_visitors.h
new file mode 100644
index 00000000000..405567e1054
--- /dev/null
+++ b/src/mongo/db/exec/sbe/stages/stage_visitors.h
@@ -0,0 +1,101 @@
+/**
+ *    Copyright (C) 2022-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/db/exec/sbe/stages/ix_scan.h"
+#include "mongo/db/query/plan_explainer_sbe.h"
+
+namespace mongo::sbe {
+/**
+ * A stage visitor that accumulates the storage-related statistics inside itself when used for
+ * walking over the SBE AST.
+ */
+class StorageAccessStatsVisitor : public PlanStageVisitor {
+public:
+    /**
+     * Adds the collected storage-related statistics into the given 'bob'.
+     */
+    BSONObj toBSON() {
+        BSONObjBuilder bob;
+        bob.appendNumber("totalDocsExamined", totalDocsExamined);
+        bob.appendNumber("totalKeysExamined", totalKeysExamined);
+        bob.appendNumber("collectionScans", collectionScans);
+        bob.appendNumber("collectionSeeks", collectionSeeks);
+        bob.appendNumber("indexScans", indexScans);
+        bob.appendNumber("indexSeeks", indexSeeks);
+        bob.append("indexesUsed", indexesUsed);
+        return bob.obj();
+    }
+
+    /**
+     * Collects the storage-related statictics for the given 'root' and 'rootStats'.
+     */
+    static StorageAccessStatsVisitor collectStats(const PlanStage& root,
+                                                  const PlanStageStats* rootStats) {
+        StorageAccessStatsVisitor res;
+        root.accumulate(res);
+        auto joinSummary = sbe::collectExecutionStatsSummary(rootStats);
+        res.totalDocsExamined = static_cast<long long>(joinSummary.totalDocsExamined);
+        res.totalKeysExamined = static_cast<long long>(joinSummary.totalKeysExamined);
+        return res;
+    }
+
+protected:
+    virtual void visit(const sbe::PlanStage* root) {
+        auto stats = root->getCommonStats();
+        if (stats->stageType == "seek"_sd) {
+            collectionSeeks += stats->opens;
+        } else if (stats->stageType == "scan"_sd) {
+            collectionScans += stats->opens;
+        } else if (stats->stageType == "ixseek"_sd || stats->stageType == "ixscan"_sd) {
+            // It's not possible to use a 'dynamic_cast' here, as 'IndexScanStage' is defined inside
+            // a separate compilation unit and it's type information is not available at this point.
+            auto indexScanStage = static_cast<const IndexScanStage*>(root);
+            uassert(6267647,
+                    str::stream() << stats->stageType
+                                  << " stage is not an instance of IndexScanStage",
+                    indexScanStage);
+            indexesUsed.push_back(indexScanStage->getIndexName());
+            if (stats->stageType == "ixseek"_sd) {
+                indexSeeks += stats->opens;
+            } else if (stats->stageType == "ixscan"_sd) {
+                indexScans += stats->opens;
+            }
+        }
+    }
+
+private:
+    long long totalDocsExamined = 0;
+    long long totalKeysExamined = 0;
+    long long collectionScans = 0;
+    long long collectionSeeks = 0;
+    long long indexScans = 0;
+    long long indexSeeks = 0;
+    std::vector<std::string> indexesUsed;
+};
+}  // namespace mongo::sbe
diff --git a/src/mongo/db/exec/sbe/stages/stages.h b/src/mongo/db/exec/sbe/stages/stages.h
index 5fdfb5650e0..e32aab56a84 100644
--- a/src/mongo/db/exec/sbe/stages/stages.h
+++ b/src/mongo/db/exec/sbe/stages/stages.h
@@ -194,6 +194,25 @@ protected:
 #endif
 };
 
+template <typename T>
+class CanTrackStats;
+
+/**
+ * An abstract class to be used for traversing a plan-stage tree.
+ */
+class PlanStageVisitor {
+public:
+    virtual ~PlanStageVisitor() = default;
+
+    friend class CanTrackStats<PlanStage>;
+
+protected:
+    /**
+     * Visits one plan-stage during a traversal over the plan-stage tree.
+     */
+    virtual void visit(const PlanStage* stage) = 0;
+};
+
 /**
  * Provides methods to obtain execution statistics specific to a plan stage.
  *
@@ -253,6 +272,18 @@ public:
         }
     }
 
+    /**
+     * Implements a pre-order traversal over the plan-stage tree starting from this node. The
+     * visitor parameter plays the role of an accumulator during this traversal.
+     */
+    void accumulate(PlanStageVisitor& visitor) const {
+        auto stage = static_cast<const T*>(this);
+        visitor.visit(stage);
+        for (auto&& child : stage->_children) {
+            child->accumulate(visitor);
+        }
+    }
+
     void detachFromTrialRunTracker() {
         auto stage = static_cast<T*>(this);
         for (auto&& child : stage->_children) {
diff --git a/src/mongo/db/exec/sbe/util/stage_results_printer.cpp b/src/mongo/db/exec/sbe/util/stage_results_printer.cpp
index 603b00ef5a0..94d2f6f3886 100644
--- a/src/mongo/db/exec/sbe/util/stage_results_printer.cpp
+++ b/src/mongo/db/exec/sbe/util/stage_results_printer.cpp
@@ -103,6 +103,43 @@ void StageResultsPrinter<T>::printSlotNames(const SlotNames& slotNames) {
     _stream << "]";
 }
 
+/**
+ * Visitor for printing the specific stats of a query stage into a stream.
+ */
+template <typename T>
+struct PlanStatsSpecificStatsPrinter : PlanStatsVisitorBase<true> {
+    PlanStatsSpecificStatsPrinter() = delete;
+    PlanStatsSpecificStatsPrinter(T& stream) : _stream(stream) {}
+
+    // To avoid overloaded-virtual warnings.
+    using PlanStatsConstVisitor::visit;
+
+    void visit(tree_walker::MaybeConstPtr<true, sbe::HashLookupStats> stats) override final {
+        _stream << "dsk:" << stats->usedDisk << "\n";
+        _stream << "htRecs:" << stats->spilledHtRecords << "\n";
+        _stream << "htIndices:" << stats->spilledHtBytesOverAllRecords << "\n";
+        _stream << "buffRecs:" << stats->spilledBuffRecords << "\n";
+        _stream << "buffBytes:" << stats->spilledBuffBytesOverAllRecords << "\n";
+    }
+
+    // TODO: Add an overload for the specific stats of other stages by overriding their
+    // corresponding `visit` method if needed.
+
+private:
+    T& _stream;
+};
+
+template <typename T>
+void StageResultsPrinter<T>::printSpecificStats(const SpecificStats* stats) {
+    auto visitor = PlanStatsSpecificStatsPrinter<T>(_stream);
+    stats->acceptVisitor(&visitor);
+}
+
+template <typename T>
+void StageResultsPrinter<T>::printSpecificStats(const PlanStage* stage) {
+    printSpecificStats(stage->getSpecificStats());
+}
+
 template class StageResultsPrinter<std::ostream>;
 template class StageResultsPrinter<str::stream>;
 
diff --git a/src/mongo/db/exec/sbe/util/stage_results_printer.h b/src/mongo/db/exec/sbe/util/stage_results_printer.h
index a96fb8e195a..7518ccf3a34 100644
--- a/src/mongo/db/exec/sbe/util/stage_results_printer.h
+++ b/src/mongo/db/exec/sbe/util/stage_results_printer.h
@@ -78,6 +78,10 @@ public:
 
     void printSlotNames(const SlotNames& slotNames);
 
+    void printSpecificStats(const PlanStage* stage);
+
+    void printSpecificStats(const SpecificStats* stats);
+
 private:
     T& _stream;
     const PrintOptions& _options;
diff --git a/src/mongo/db/query/plan_explainer_sbe.cpp b/src/mongo/db/query/plan_explainer_sbe.cpp
index b18e6cdec12..2f8f5b43b08 100644
--- a/src/mongo/db/query/plan_explainer_sbe.cpp
+++ b/src/mongo/db/query/plan_explainer_sbe.cpp
@@ -301,22 +301,6 @@ void statsToBSON(const sbe::PlanStageStats* stats,
     statsToBSONHelper(stats, bob, topLevelBob, 0);
 }
 
-PlanSummaryStats collectExecutionStatsSummary(const sbe::PlanStageStats* stats) {
-    invariant(stats);
-
-    PlanSummaryStats summary;
-    summary.nReturned = stats->common.advances;
-
-    if (stats->common.executionTimeMillis) {
-        summary.executionTimeMillisEstimate = *stats->common.executionTimeMillis;
-    }
-
-    auto visitor = PlanSummaryStatsVisitor(summary);
-    auto walker = PlanStageStatsWalker<true, sbe::CommonStats>(nullptr, nullptr, &visitor);
-    tree_walker::walk<true, sbe::PlanStageStats>(stats, &walker);
-    return summary;
-}
-
 PlanExplainer::PlanStatsDetails buildPlanStatsDetails(
     const QuerySolution* solution,
     const sbe::PlanStageStats* stats,
@@ -326,7 +310,7 @@ PlanExplainer::PlanStatsDetails buildPlanStatsDetails(
     BSONObjBuilder bob;
 
     if (verbosity >= ExplainOptions::Verbosity::kExecStats) {
-        auto summary = collectExecutionStatsSummary(stats);
+        auto summary = sbe::collectExecutionStatsSummary(stats);
         if (solution != nullptr && verbosity >= ExplainOptions::Verbosity::kExecAllPlans) {
             summary.score = solution->score;
         }
diff --git a/src/mongo/db/test_output/exec/sbe/hash_lookup_stage_test/basic_tests.txt b/src/mongo/db/test_output/exec/sbe/hash_lookup_stage_test/basic_tests.txt
index edaff3d434d..035f848784f 100644
--- a/src/mongo/db/test_output/exec/sbe/hash_lookup_stage_test/basic_tests.txt
+++ b/src/mongo/db/test_output/exec/sbe/hash_lookup_stage_test/basic_tests.txt
@@ -12,6 +12,37 @@ INNER [value, key]:
 {"_id" : 13}, 2
 {"_id" : 14}, 7
 
+--- First Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Second Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Third Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Fourth Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Fifth Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+
 -- OUTPUT [outer_key, inner_agg]:
 1, [{"_id" : 11}]
 2, [{"_id" : 12}, {"_id" : 13}]
@@ -28,6 +59,37 @@ INNER [value, key]:
 {"_id" : 11}, 2
 {"_id" : 12}, 3
 
+--- First Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Second Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Third Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Fourth Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Fifth Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+
 -- OUTPUT [outer_key, inner_agg]:
 1, Nothing
 4, Nothing
@@ -46,6 +108,37 @@ INNER [value, key]:
 {"_id" : 13}, [1, 2, 3]
 {"_id" : 14}, []
 
+--- First Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Second Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Third Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Fourth Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Fifth Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+
 -- OUTPUT [outer_key, inner_agg]:
 1, [{"_id" : 13}]
 [2, 3], [{"_id" : 11}, {"_id" : 13}]
@@ -72,6 +165,37 @@ INNER [value, key]:
 {"_id" : 16}, [[3]]
 {"_id" : 17}, [2, [3]]
 
+--- First Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Second Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Third Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Fourth Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Fifth Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+
 -- OUTPUT [outer_key, inner_agg]:
 1, [{"_id" : 11}, {"_id" : 14}]
 2, [{"_id" : 12}, {"_id" : 17}]
@@ -103,6 +227,37 @@ INNER [value, key]:
 {"_id" : 17}, [{"a" : 1, "b" : 1}]
 {"_id" : 18}, [{"b" : 1, "a" : 1}]
 
+--- First Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Second Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Third Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Fourth Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Fifth Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+
 -- OUTPUT [outer_key, inner_agg]:
 {"a" : 1}, [{"_id" : 11}, {"_id" : 15}, {"_id" : 16}]
 {"b" : 1}, [{"_id" : 12}, {"_id" : 16}]
@@ -137,6 +292,37 @@ INNER [value, key]:
 {"_id" : 18}, [null, "1", "abc", NumberDecimal(1)]
 {"_id" : 19}, [{"x" : 1, "y" : "abc"}]
 
+--- First Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Second Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Third Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Fourth Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Fifth Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+
 -- OUTPUT [outer_key, inner_agg]:
 null, [{"_id" : 11}, {"_id" : 17}, {"_id" : 18}]
 1, [{"_id" : 12}, {"_id" : 14}, {"_id" : 15}, {"_id" : 18}]
@@ -165,6 +351,37 @@ INNER [value, key]:
 {"_id" : 14}, "Abc"
 {"_id" : 15}, "def"
 
+--- First Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Second Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Third Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Fourth Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Fifth Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+
 -- OUTPUT [outer_key, inner_agg]:
 null, [{"_id" : 11}]
 "abc", [{"_id" : 12}, {"_id" : 13}, {"_id" : 14}]
@@ -178,6 +395,37 @@ OUTER [value, key]:
 
 INNER [value, key]:
 
+--- First Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Second Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Third Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Fourth Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Fifth Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+
 -- OUTPUT [outer_key, inner_agg]:
 
 ==== VARIATION: empty outer ====
@@ -188,6 +436,37 @@ INNER [value, key]:
 {"_id" : 11}, 2
 {"_id" : 12}, 3
 
+--- First Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Second Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Third Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Fourth Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Fifth Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+
 -- OUTPUT [outer_key, inner_agg]:
 
 ==== VARIATION: empty inner ====
@@ -198,6 +477,37 @@ OUTER [value, key]:
 
 INNER [value, key]:
 
+--- First Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Second Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Third Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Fourth Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+--- Fifth Stats
+dsk:0
+htRecs:0
+htIndices:0
+buffRecs:0
+buffBytes:0
+
 -- OUTPUT [outer_key, inner_agg]:
 1, Nothing
 2, Nothing