SERVER-64347 Add support for descending sort to the bounded sorter

6 files changed, 398 insertions, 165 deletions

diff --git a/jstests/core/timeseries/timeseries_internal_bounded_sort.js b/jstests/core/timeseries/timeseries_internal_bounded_sort.js
index 0c113e1eab1..f16ae458143 100644
--- a/jstests/core/timeseries/timeseries_internal_bounded_sort.js
+++ b/jstests/core/timeseries/timeseries_internal_bounded_sort.js
@@ -51,71 +51,85 @@ assert.commandWorked(coll.createIndex({t: 1}));
 
 const unpackStage = getAggPlanStage(coll.explain().aggregate(), '$_internalUnpackBucket');
 
-function assertSorted(result) {
-    let prev = {t: -Infinity};
+function assertSorted(result, ascending) {
+    let prev = ascending ? {t: -Infinity} : {t: Infinity};
     for (const doc of result) {
-        assert.lte(+prev.t, +doc.t, 'Found two docs not in time order: ' + tojson({prev, doc}));
+        if (ascending) {
+            assert.lte(+prev.t,
+                       +doc.t,
+                       'Found two docs not in ascending time order: ' + tojson({prev, doc}));
+        } else {
+            assert.gte(+prev.t,
+                       +doc.t,
+                       'Found two docs not in descending time order: ' + tojson({prev, doc}));
+        }
+
         prev = doc;
     }
 }
 
-// Test sorting the whole collection.
-{
-    const naive = buckets
-                      .aggregate([
-                          unpackStage,
-                          {$_internalInhibitOptimization: {}},
-                          {$sort: {t: 1}},
-                      ])
-                      .toArray();
-    assertSorted(naive);
+function runTest(ascending) {
+    // Test sorting the whole collection.
+    {
+        const naive = buckets
+                          .aggregate([
+                              unpackStage,
+                              {$_internalInhibitOptimization: {}},
+                              {$sort: {t: ascending ? 1 : -1}},
+                          ])
+                          .toArray();
+        assertSorted(naive, ascending);
 
-    const opt = buckets
-                    .aggregate([
-                        {$sort: {'control.min.t': 1}},
-                        unpackStage,
-                        {
-                            $_internalBoundedSort: {
-                                sortKey: {t: 1},
-                                bound: bucketMaxSpanSeconds,
-                            }
-                        },
-                    ])
-                    .toArray();
-    assertSorted(opt);
+        const opt = buckets
+                        .aggregate([
+                            {$sort: {'control.min.t': ascending ? 1 : -1}},
+                            unpackStage,
+                            {
+                                $_internalBoundedSort: {
+                                    sortKey: {t: ascending ? 1 : -1},
+                                    bound: bucketMaxSpanSeconds,
+                                }
+                            },
+                        ])
+                        .toArray();
+        assertSorted(opt, ascending);
 
-    assert.eq(naive, opt);
-}
+        assert.eq(naive, opt);
+    }
 
-// Test $sort + $limit.
-{
-    const naive = buckets
-                      .aggregate([
-                          unpackStage,
-                          {$_internalInhibitOptimization: {}},
-                          {$sort: {t: 1}},
-                          {$limit: 100},
-                      ])
-                      .toArray();
-    assertSorted(naive);
-    assert.eq(100, naive.length);
+    // Test $sort + $limit.
+    {
+        const naive = buckets
+                          .aggregate([
+                              unpackStage,
+                              {$_internalInhibitOptimization: {}},
+                              {$sort: {t: ascending ? 1 : -1}},
+                              {$limit: 100},
+                          ])
+                          .toArray();
+        assertSorted(naive, ascending);
+        assert.eq(100, naive.length);
 
-    const opt = buckets
-                    .aggregate([
-                        {$sort: {'control.min.t': 1}},
-                        unpackStage,
-                        {
-                            $_internalBoundedSort: {
-                                sortKey: {t: 1},
-                                bound: bucketMaxSpanSeconds,
-                            }
-                        },
-                        {$limit: 100},
-                    ])
-                    .toArray();
-    assertSorted(opt);
-    assert.eq(100, opt.length);
+        const opt = buckets
+                        .aggregate([
+                            {$sort: {'control.min.t': ascending ? 1 : -1}},
+                            unpackStage,
+                            {
+                                $_internalBoundedSort: {
+                                    sortKey: {t: ascending ? 1 : -1},
+                                    bound: bucketMaxSpanSeconds,
+                                }
+                            },
+                            {$limit: 100},
+                        ])
+                        .toArray();
+        assertSorted(opt, ascending);
+        assert.eq(100, opt.length);
 
-    assert.eq(naive, opt);
+        assert.eq(naive, opt);
+    }
 }
+
+runTest(true);   // ascending
+runTest(false);  // descending
 })();
diff --git a/src/mongo/db/pipeline/document_source_sort.cpp b/src/mongo/db/pipeline/document_source_sort.cpp
index 2e341acc6c3..b5ef461e697 100644
--- a/src/mongo/db/pipeline/document_source_sort.cpp
+++ b/src/mongo/db/pipeline/document_source_sort.cpp
@@ -57,6 +57,57 @@ using std::string;
 using std::unique_ptr;
 using std::vector;
 
+namespace {
+struct BoundMakerAsc {
+    const Seconds bound;
+
+    DocumentSourceSort::SortableDate operator()(DocumentSourceSort::SortableDate key) const {
+        return {key.date - bound};
+    }
+
+    long long serialize() const {
+        return duration_cast<Seconds>(bound).count();
+    }
+};
+
+struct BoundMakerDesc {
+    const Seconds bound;
+
+    DocumentSourceSort::SortableDate operator()(DocumentSourceSort::SortableDate key) const {
+        return {key.date + bound};
+    }
+
+    long long serialize() const {
+        return duration_cast<Seconds>(bound).count();
+    }
+};
+struct CompAsc {
+    int operator()(DocumentSourceSort::SortableDate x, DocumentSourceSort::SortableDate y) const {
+        // compare(x, y) op 0 means x op y, for any comparator 'op'.
+        if (x.date.toMillisSinceEpoch() < y.date.toMillisSinceEpoch())
+            return -1;
+        if (x.date.toMillisSinceEpoch() > y.date.toMillisSinceEpoch())
+            return 1;
+        return 0;
+    }
+};
+struct CompDesc {
+    int operator()(DocumentSourceSort::SortableDate x, DocumentSourceSort::SortableDate y) const {
+        // compare(x, y) op 0 means x op y, for any comparator 'op'.
+        if (x.date.toMillisSinceEpoch() > y.date.toMillisSinceEpoch())
+            return -1;
+        if (x.date.toMillisSinceEpoch() < y.date.toMillisSinceEpoch())
+            return 1;
+        return 0;
+    }
+};
+
+using TimeSorterAsc =
+    BoundedSorter<DocumentSourceSort::SortableDate, Document, CompAsc, BoundMakerAsc>;
+using TimeSorterDesc =
+    BoundedSorter<DocumentSourceSort::SortableDate, Document, CompDesc, BoundMakerDesc>;
+}  // namespace
+
 constexpr StringData DocumentSourceSort::kStageName;
 
 DocumentSourceSort::DocumentSourceSort(const boost::intrusive_ptr<ExpressionContext>& pExpCtx,
@@ -92,7 +143,7 @@ REGISTER_DOCUMENT_SOURCE_CONDITIONALLY(
 DocumentSource::GetNextResult DocumentSourceSort::doGetNext() {
     if (_timeSorter) {
         // Only pull input as necessary to get _timeSorter to have a result.
-        while (_timeSorter->getState() == TimeSorter::State::kWait) {
+        while (_timeSorter->getState() == TimeSorterInterface::State::kWait) {
             auto input = pSource->getNext();
             switch (input.getStatus()) {
                 case GetNextResult::ReturnStatus::kPauseExecution:
@@ -113,7 +164,7 @@ DocumentSource::GetNextResult DocumentSourceSort::doGetNext() {
             }
         }
 
-        if (_timeSorter->getState() == TimeSorter::State::kDone)
+        if (_timeSorter->getState() == TimeSorterInterface::State::kDone)
             return GetNextResult::makeEOF();
 
         return _timeSorter->next().second;
@@ -157,7 +208,7 @@ void DocumentSourceSort::serializeToArray(
             {"$_internalBoundedSort"_sd,
              Document{{
                  {"sortKey"_sd, std::move(sortKey)},
-                 {"bound"_sd, durationCount<Seconds>(_timeSorter->makeBound.bound)},
+                 {"bound"_sd, _timeSorter->serializeBound()},
              }}},
         }}});
         return;
@@ -286,7 +337,6 @@ intrusive_ptr<DocumentSourceSort> DocumentSourceSort::parseBoundedSort(
     uassert(6369904, "$_internalBoundedSort sortKey must be an object", key.type() == Object);
     SortPattern pat{key.embeddedObject(), expCtx};
     uassert(6369903, "$_internalBoundedSort doesn't support compound sort", pat.size() == 1);
-    uassert(6369902, "$_internalBoundedSort only handles ascending sort", pat[0].isAscending);
     uassert(6369901,
             "$_internalBoundedSort doesn't support an expression in the sortKey",
             pat[0].expression == nullptr);
@@ -305,7 +355,12 @@ intrusive_ptr<DocumentSourceSort> DocumentSourceSort::parseBoundedSort(
     }
 
     auto ds = DocumentSourceSort::create(expCtx, pat);
-    ds->_timeSorter.reset(new TimeSorter{opts, {}, BoundMaker{Seconds{boundN}}});
+    if (pat[0].isAscending) {
+        ds->_timeSorter.reset(new TimeSorterAsc{opts, CompAsc{}, BoundMakerAsc{Seconds{boundN}}});
+    } else {
+        ds->_timeSorter.reset(
+            new TimeSorterDesc{opts, CompDesc{}, BoundMakerDesc{Seconds{boundN}}});
+    }
     return ds;
 }
 
@@ -407,5 +462,9 @@ std::string nextFileName() {
 #include "mongo/db/sorter/sorter.cpp"
 template class ::mongo::BoundedSorter<::mongo::DocumentSourceSort::SortableDate,
                                       ::mongo::Document,
-                                      ::mongo::DocumentSourceSort::Comp,
-                                      ::mongo::DocumentSourceSort::BoundMaker>;
+                                      ::mongo::CompAsc,
+                                      ::mongo::BoundMakerAsc>;
+template class ::mongo::BoundedSorter<::mongo::DocumentSourceSort::SortableDate,
+                                      ::mongo::Document,
+                                      ::mongo::CompDesc,
+                                      ::mongo::BoundMakerDesc>;
diff --git a/src/mongo/db/pipeline/document_source_sort.h b/src/mongo/db/pipeline/document_source_sort.h
index 4241854878b..cd669051e67 100644
--- a/src/mongo/db/pipeline/document_source_sort.h
+++ b/src/mongo/db/pipeline/document_source_sort.h
@@ -42,6 +42,21 @@ namespace mongo {
 
 class DocumentSourceSort final : public DocumentSource {
 public:
+    struct SortableDate {
+        Date_t date;
+
+        struct SorterDeserializeSettings {};  // unused
+        void serializeForSorter(BufBuilder& buf) const {
+            buf.appendNum(date.toMillisSinceEpoch());
+        }
+        static SortableDate deserializeForSorter(BufReader& buf, const SorterDeserializeSettings&) {
+            return {Date_t::fromMillisSinceEpoch(buf.read<LittleEndian<long long>>().value)};
+        }
+        int memUsageForSorter() const {
+            return sizeof(SortableDate);
+        }
+    };
+
     static constexpr StringData kStageName = "$sort"_sd;
 
     const char* getSourceName() const final {
@@ -196,40 +211,8 @@ private:
 
     boost::optional<SortKeyGenerator> _sortKeyGen;
 
-    struct SortableDate {
-        Date_t date;
-
-        struct SorterDeserializeSettings {};  // unused
-        void serializeForSorter(BufBuilder& buf) const {
-            buf.appendNum(date.toMillisSinceEpoch());
-        }
-        static SortableDate deserializeForSorter(BufReader& buf, const SorterDeserializeSettings&) {
-            return {Date_t::fromMillisSinceEpoch(buf.read<LittleEndian<long long>>().value)};
-        }
-        int memUsageForSorter() const {
-            return sizeof(SortableDate);
-        }
-    };
-
-    struct BoundMaker {
-        Seconds bound;
-
-        SortableDate operator()(SortableDate key) {
-            return {key.date - bound};
-        }
-    };
-    struct Comp {
-        int operator()(SortableDate x, SortableDate y) const {
-            // compare(x, y) op 0 means x op y, for any comparator 'op'.
-            if (x.date.toMillisSinceEpoch() < y.date.toMillisSinceEpoch())
-                return -1;
-            if (x.date.toMillisSinceEpoch() > y.date.toMillisSinceEpoch())
-                return 1;
-            return 0;
-        }
-    };
-    using TimeSorter = BoundedSorter<SortableDate, Document, Comp, BoundMaker>;
-    std::unique_ptr<TimeSorter> _timeSorter;
+    using TimeSorterInterface = BoundedSorterInterface<SortableDate, Document>;
+    std::unique_ptr<TimeSorterInterface> _timeSorter;
 };
 
 }  // namespace mongo
diff --git a/src/mongo/db/sorter/sorter.cpp b/src/mongo/db/sorter/sorter.cpp
index e520d6d4c23..65eac71aba4 100644
--- a/src/mongo/db/sorter/sorter.cpp
+++ b/src/mongo/db/sorter/sorter.cpp
@@ -1350,14 +1350,14 @@ SortIteratorInterface<Key, Value>* SortedFileWriter<Key, Value>::done() {
 template <typename Key, typename Value, typename Comparator, typename BoundMaker>
 BoundedSorter<Key, Value, Comparator, BoundMaker>::BoundedSorter(const SortOptions& opts,
                                                                  Comparator comp,
-                                                                 BoundMaker makeBound)
+                                                                 BoundMaker makeBound,
+                                                                 bool checkInput)
     : compare(comp),
       makeBound(makeBound),
-      _comparePairs([this](const std::pair<Key, Value>& lhs, const std::pair<Key, Value>& rhs) {
-          return this->compare(lhs.first, rhs.first);
-      }),
+      _comparePairs{compare},
+      _checkInput(checkInput),
       _opts(opts),
-      _heap(Greater{comp}),
+      _heap(Greater{&compare}),
       _file(opts.extSortAllowed ? std::make_shared<typename Sorter<Key, Value>::File>(
                                       opts.tempDir + "/" + nextFileName())
                                 : nullptr) {}
@@ -1368,7 +1368,7 @@ void BoundedSorter<Key, Value, Comparator, BoundMaker>::add(Key key, Value value
     // If a new value violates what we thought was our min bound, something has gone wrong.
     uassert(6369910,
             "BoundedSorter input is too out-of-order.",
-            !checkInput || !_min || compare(*_min, key) <= 0);
+            !_checkInput || !_min || compare(*_min, key) <= 0);
 
     // Each new item can potentially give us a tighter bound (a higher min).
     Key newMin = makeBound(key);
@@ -1386,7 +1386,7 @@ void BoundedSorter<Key, Value, Comparator, BoundMaker>::add(Key key, Value value
 }
 
 template <typename Key, typename Value, typename Comparator, typename BoundMaker>
-typename BoundedSorter<Key, Value, Comparator, BoundMaker>::State
+typename BoundedSorterInterface<Key, Value>::State
 BoundedSorter<Key, Value, Comparator, BoundMaker>::getState() const {
     if (_opts.limit > 0 && _opts.limit == _numSorted) {
         return State::kDone;
@@ -1459,11 +1459,11 @@ void BoundedSorter<Key, Value, Comparator, BoundMaker>::_spill() {
     if (_heap.empty())
         return;
 
-    // If we have a small $limit, we can simply extract that many of the smallest elements from the
-    // _heap and discard the rest, avoiding an expensive spill to disk.
+    // If we have a small $limit, we can simply extract that many of the smallest elements from
+    // the _heap and discard the rest, avoiding an expensive spill to disk.
     if (_opts.limit > 0 && _opts.limit < (_heap.size() / 2)) {
         _memUsed = 0;
-        decltype(_heap) retained;
+        decltype(_heap) retained{Greater{&compare}};
         for (size_t i = 0; i < _opts.limit; ++i) {
             _memUsed +=
                 _heap.top().first.memUsageForSorter() + _heap.top().second.memUsageForSorter();
@@ -1505,6 +1505,12 @@ void BoundedSorter<Key, Value, Comparator, BoundMaker>::_spill() {
     _memUsed = 0;
 }
 
+template <typename Key, typename Value, typename Comparator, typename BoundMaker>
+int BoundedSorter<Key, Value, Comparator, BoundMaker>::PairComparator::operator()(
+    const std::pair<Key, Value>& p1, const std::pair<Key, Value>& p2) const {
+    return compare(p1.first, p2.first);
+}
+
 //
 // Factory Functions
 //
diff --git a/src/mongo/db/sorter/sorter.h b/src/mongo/db/sorter/sorter.h
index fc66cea77c2..d628777ad8b 100644
--- a/src/mongo/db/sorter/sorter.h
+++ b/src/mongo/db/sorter/sorter.h
@@ -372,6 +372,48 @@ protected:
     std::vector<std::shared_ptr<Iterator>> _iters;  // Data that has already been spilled.
 };
 
+
+template <typename Key, typename Value>
+class BoundedSorterInterface {
+public:
+    virtual ~BoundedSorterInterface() {}
+
+    // Feed one item of input to the sorter.
+    // Together, add() and done() represent the input stream.
+    virtual void add(Key key, Value value) = 0;
+
+    // Indicate that no more input will arrive.
+    // Together, add() and done() represent the input stream.
+    virtual void done() = 0;
+
+    enum class State {
+        // An output document is not available yet, but this may change as more input arrives.
+        kWait,
+        // An output document is available now: you may call next() once.
+        kReady,
+        // All output has been returned.
+        kDone,
+    };
+    // Together, state() and next() represent the output stream.
+    // See BoundedSorter::State for the meaning of each case.
+    virtual State getState() const = 0;
+
+    // Remove and return one item of output.
+    // Only valid to call when getState() == kReady.
+    // Together, state() and next() represent the output stream.
+    virtual std::pair<Key, Value> next() = 0;
+
+    // Serialize the bound for explain output
+    virtual long long serializeBound() const = 0;
+
+    virtual size_t numSpills() const = 0;
+
+    // By default, uassert that the input meets our assumptions of being almost-sorted.
+    // But if _checkInput is false, don't do that check.
+    // The output will be in the wrong order but otherwise it should work.
+    virtual bool checkInput() const = 0;
+};
+
 /**
  * Sorts data that is already "almost sorted", meaning we can put a bound on how out-of-order
  * any two input elements are. For example, maybe we are sorting by {time: 1} and we know that no
@@ -391,7 +433,7 @@ protected:
  * less-or-equal to all future Keys that will be seen in the input.
  */
 template <typename Key, typename Value, typename Comparator, typename BoundMaker>
-class BoundedSorter {
+class BoundedSorter : public BoundedSorterInterface<Key, Value> {
 public:
     // 'Comparator' is a 3-way comparison, but std::priority_queue wants a '<' comparison.
     // But also, std::priority_queue is a max-heap, and we want a min-heap.
@@ -399,12 +441,20 @@ public:
     // on whole elements.
     struct Greater {
         bool operator()(const std::pair<Key, Value>& p1, const std::pair<Key, Value>& p2) const {
-            return compare(p1.first, p2.first) > 0;
+            return (*compare)(p1.first, p2.first) > 0;
         }
-        Comparator compare;
+        Comparator const* compare;
     };
 
-    BoundedSorter(const SortOptions& opts, Comparator comp, BoundMaker makeBound);
+    BoundedSorter(const SortOptions& opts,
+                  Comparator comp,
+                  BoundMaker makeBound,
+                  bool checkInput = true);
+
+    BoundedSorter(const BoundedSorter&) = delete;
+    BoundedSorter(BoundedSorter&&) = delete;
+    BoundedSorter& operator=(const BoundedSorter&) = delete;
+    BoundedSorter& operator=(BoundedSorter&&) = delete;
 
     // Feed one item of input to the sorter.
     // Together, add() and done() represent the input stream.
@@ -417,16 +467,9 @@ public:
         _done = true;
     }
 
-    enum class State {
-        // An output document is not available yet, but this may change as more input arrives.
-        kWait,
-        // An output document is available now: you may call next() once.
-        kReady,
-        // All output has been returned.
-        kDone,
-    };
     // Together, state() and next() represent the output stream.
     // See BoundedSorter::State for the meaning of each case.
+    using State = typename BoundedSorterInterface<Key, Value>::State;
     State getState() const;
 
     // Remove and return one item of output.
@@ -434,26 +477,34 @@ public:
     // Together, state() and next() represent the output stream.
     std::pair<Key, Value> next();
 
+    // Serialize the bound for explain output
+    long long serializeBound() const {
+        return makeBound.serialize();
+    };
+
     size_t numSpills() const {
         return _numSpills;
     }
 
-    // By default, uassert that the input meets our assumptions of being almost-sorted.
-    // But if _checkInput is false, don't do that check.
-    // The output will be in the wrong order but otherwise it should work.
-    bool checkInput = true;
+    bool checkInput() const {
+        return _checkInput;
+    }
 
-    Comparator compare;
-    BoundMaker makeBound;
+    const Comparator compare;
+    const BoundMaker makeBound;
 
 private:
     using SpillIterator = SortIteratorInterface<Key, Value>;
-    using PairComparator =
-        std::function<int(const std::pair<Key, Value>&, const std::pair<Key, Value>&)>;
+    struct PairComparator {
+        int operator()(const std::pair<Key, Value>& p1, const std::pair<Key, Value>& p2) const;
+        const Comparator& compare;
+    };
 
     void _spill();
 
-    PairComparator _comparePairs;
+    const PairComparator _comparePairs;
+
+    bool _checkInput;
 
     size_t _numSorted = 0;              // Keeps track of the number of keys sorted.
     uint64_t _totalDataSizeSorted = 0;  // Keeps track of the total size of data sorted.
diff --git a/src/mongo/db/sorter/sorter_test.cpp b/src/mongo/db/sorter/sorter_test.cpp
index 02454b2756f..edbda871916 100644
--- a/src/mongo/db/sorter/sorter_test.cpp
+++ b/src/mongo/db/sorter/sorter_test.cpp
@@ -994,17 +994,36 @@ public:
             return sizeof(Doc);
         }
     };
-    struct Comparator {
+    struct ComparatorAsc {
         int operator()(Key x, Key y) const {
             return x - y;
         }
     };
-    struct BoundMaker {
+    struct ComparatorDesc {
+        int operator()(Key x, Key y) const {
+            return y - x;
+        }
+    };
+    struct BoundMakerAsc {
         Key operator()(Key k) const {
             return k - 10;
         }
+        long long serialize() const {
+            return 10;
+        }
+    };
+    struct BoundMakerDesc {
+        Key operator()(Key k) const {
+            return k + 10;
+        }
+        long long serialize() const {
+            return 10;
+        }
     };
-    using S = BoundedSorter<Key, Doc, Comparator, BoundMaker>;
+
+    using S = BoundedSorterInterface<Key, Doc>;
+    using SAsc = BoundedSorter<Key, Doc, ComparatorAsc, BoundMakerAsc>;
+    using SDesc = BoundedSorter<Key, Doc, ComparatorDesc, BoundMakerDesc>;
 
     /**
      * Feed the input into the sorter one-by-one, taking any output as soon as it's available.
@@ -1014,35 +1033,46 @@ public:
         auto push = [&](Doc doc) { output.push_back(doc); };
 
         for (auto&& doc : input) {
-            sorter.add(doc.time, doc);
-            while (sorter.getState() == S::State::kReady)
-                push(sorter.next().second);
+            sorter->add(doc.time, doc);
+            while (sorter->getState() == S::State::kReady)
+                push(sorter->next().second);
         }
-        sorter.done();
+        sorter->done();
 
-        while (sorter.getState() == S::State::kReady)
-            push(sorter.next().second);
-        ASSERT(sorter.getState() == S::State::kDone);
+        while (sorter->getState() == S::State::kReady)
+            push(sorter->next().second);
+        ASSERT(sorter->getState() == S::State::kDone);
 
         ASSERT_EQ(output.size(), expectedSize == -1 ? input.size() : expectedSize);
         return output;
     }
 
-    static void assertSorted(const std::vector<Doc>& docs) {
+    static void assertSorted(const std::vector<Doc>& docs, bool ascending = true) {
         for (size_t i = 1; i < docs.size(); ++i) {
             Doc prev = docs[i - 1];
             Doc curr = docs[i];
-            ASSERT_LTE(prev.time, curr.time);
+            if (ascending) {
+                ASSERT_LTE(prev.time, curr.time);
+            } else {
+                ASSERT_GTE(prev.time, curr.time);
+            }
         }
     }
 
-    S sorter{{}, {}, {}};
+    std::unique_ptr<S> makeAsc(SortOptions options, bool checkInput = true) {
+        return std::make_unique<SAsc>(options, ComparatorAsc{}, BoundMakerAsc{}, checkInput);
+    }
+    std::unique_ptr<S> makeDesc(SortOptions options, bool checkInput = true) {
+        return std::make_unique<SDesc>(options, ComparatorDesc{}, BoundMakerDesc{}, checkInput);
+    }
+
+    std::unique_ptr<S> sorter = makeAsc({});
 };
 TEST_F(BoundedSorterTest, Empty) {
-    ASSERT(sorter.getState() == S::State::kWait);
+    ASSERT(sorter->getState() == S::State::kWait);
 
-    sorter.done();
-    ASSERT(sorter.getState() == S::State::kDone);
+    sorter->done();
+    ASSERT(sorter->getState() == S::State::kDone);
 }
 TEST_F(BoundedSorterTest, Sorted) {
     auto output = sort({
@@ -1108,7 +1138,7 @@ TEST_F(BoundedSorterTest, WrongInput) {
     };
 
     // Disable input order checking so we can see what happens.
-    sorter.checkInput = false;
+    sorter = makeAsc({}, /* checkInput */ false);
     auto output = sort(input);
     ASSERT_EQ(output.size(), 7);
 
@@ -1121,14 +1151,14 @@ TEST_F(BoundedSorterTest, WrongInput) {
     ASSERT_EQ(output[6].time, 16);
 
     // Test that by default, bad input like this would be detected.
-    sorter = S{{}, {}, {}};
-    ASSERT(sorter.checkInput);
+    sorter = makeAsc({});
+    ASSERT(sorter->checkInput());
     ASSERT_THROWS_CODE(sort(input), DBException, 6369910);
 }
 
 TEST_F(BoundedSorterTest, MemoryLimitsNoExtSortAllowed) {
     auto options = SortOptions().MaxMemoryUsageBytes(16);
-    sorter = S(options, {}, {});
+    sorter = makeAsc(options);
 
     std::vector<Doc> input = {
         {0},
@@ -1149,7 +1179,7 @@ TEST_F(BoundedSorterTest, MemoryLimitsNoExtSortAllowed) {
 TEST_F(BoundedSorterTest, SpillSorted) {
     auto options =
         SortOptions().ExtSortAllowed().TempDir("unused_temp_dir").MaxMemoryUsageBytes(16);
-    sorter = S(options, {}, {});
+    sorter = makeAsc(options);
 
     auto output = sort({
         {0},
@@ -1164,13 +1194,13 @@ TEST_F(BoundedSorterTest, SpillSorted) {
     });
     assertSorted(output);
 
-    ASSERT_EQ(sorter.numSpills(), 3);
+    ASSERT_EQ(sorter->numSpills(), 3);
 }
 
 TEST_F(BoundedSorterTest, SpillSortedExceptOne) {
     auto options =
         SortOptions().ExtSortAllowed().TempDir("unused_temp_dir").MaxMemoryUsageBytes(16);
-    sorter = S(options, {}, {});
+    sorter = makeAsc(options);
 
     auto output = sort({
         {0},
@@ -1186,13 +1216,13 @@ TEST_F(BoundedSorterTest, SpillSortedExceptOne) {
     });
     assertSorted(output);
 
-    ASSERT_EQ(sorter.numSpills(), 3);
+    ASSERT_EQ(sorter->numSpills(), 3);
 }
 
 TEST_F(BoundedSorterTest, SpillAlmostSorted) {
     auto options =
         SortOptions().ExtSortAllowed().TempDir("unused_temp_dir").MaxMemoryUsageBytes(16);
-    sorter = S(options, {}, {});
+    sorter = makeAsc(options);
 
     auto output = sort({
         // 0 and 11 cannot swap.
@@ -1209,13 +1239,12 @@ TEST_F(BoundedSorterTest, SpillAlmostSorted) {
     });
     assertSorted(output);
 
-    ASSERT_EQ(sorter.numSpills(), 2);
+    ASSERT_EQ(sorter->numSpills(), 2);
 }
 
 TEST_F(BoundedSorterTest, SpillWrongInput) {
     auto options =
         SortOptions().ExtSortAllowed().TempDir("unused_temp_dir").MaxMemoryUsageBytes(16);
-    sorter = S(options, {}, {});
 
     std::vector<Doc> input = {
         {3},
@@ -1232,7 +1261,7 @@ TEST_F(BoundedSorterTest, SpillWrongInput) {
     };
 
     // Disable input order checking so we can see what happens.
-    sorter.checkInput = false;
+    sorter = makeAsc(options, /* checkInput */ false);
     auto output = sort(input);
     ASSERT_EQ(output.size(), 7);
 
@@ -1244,18 +1273,18 @@ TEST_F(BoundedSorterTest, SpillWrongInput) {
     ASSERT_EQ(output[5].time, 15);
     ASSERT_EQ(output[6].time, 16);
 
-    ASSERT_EQ(sorter.numSpills(), 2);
+    ASSERT_EQ(sorter->numSpills(), 2);
 
     // Test that by default, bad input like this would be detected.
-    sorter = S{options, {}, {}};
-    ASSERT(sorter.checkInput);
+    sorter = makeAsc(options);
+    ASSERT(sorter->checkInput());
     ASSERT_THROWS_CODE(sort(input), DBException, 6369910);
 }
 
 TEST_F(BoundedSorterTest, LimitNoSpill) {
     auto options =
         SortOptions().ExtSortAllowed().TempDir("unused_temp_dir").MaxMemoryUsageBytes(40).Limit(2);
-    sorter = S(options, {}, {});
+    sorter = makeAsc(options);
 
     auto output = sort(
         {
@@ -1274,13 +1303,13 @@ TEST_F(BoundedSorterTest, LimitNoSpill) {
         2);
     assertSorted(output);
 
-    ASSERT_EQ(sorter.numSpills(), 0);
+    ASSERT_EQ(sorter->numSpills(), 0);
 }
 
 TEST_F(BoundedSorterTest, LimitSpill) {
     auto options =
         SortOptions().ExtSortAllowed().TempDir("unused_temp_dir").MaxMemoryUsageBytes(40).Limit(3);
-    sorter = S(options, {}, {});
+    sorter = makeAsc(options);
 
     auto output = sort(
         {
@@ -1299,9 +1328,96 @@ TEST_F(BoundedSorterTest, LimitSpill) {
         3);
     assertSorted(output);
 
-    ASSERT_EQ(sorter.numSpills(), 1);
+    ASSERT_EQ(sorter->numSpills(), 1);
+}
+
+TEST_F(BoundedSorterTest, DescSorted) {
+    sorter = makeDesc({});
+    auto output = sort({
+        {16},
+        {15},
+        {14},
+        {13},
+        {12},
+        {11},
+        {10},
+        {3},
+        {0},
+    });
+    assertSorted(output, /* ascending */ false);
 }
 
+TEST_F(BoundedSorterTest, DescSortedExceptOne) {
+    sorter = makeDesc({});
+
+    auto output = sort({
+
+        {16},
+        {15},
+        {14},
+        {13},
+        // Swap 11 and 12.
+        {11},
+        {12},
+        {10},
+        {3},
+        {0},
+    });
+    assertSorted(output, /* ascending */ false);
+}
+
+TEST_F(BoundedSorterTest, DescAlmostSorted) {
+    sorter = makeDesc({});
+
+    auto output = sort({
+        {16},
+        {15},
+        // 3 and 14 cannot swap.
+        {14},
+        {3},
+        {12},
+        {10},
+        {13},
+        // 0 and 11 cannot swap.
+        {11},
+        {0},
+    });
+    assertSorted(output, /* ascending */ false);
+}
+
+TEST_F(BoundedSorterTest, DescWrongInput) {
+    std::vector<Doc> input = {
+        {16},
+        {14},
+        {10},
+        {5},
+        {3},
+        // This 15 is too far out of order: it's more than 10 away from 3.
+        // So it will appear too late in the output.
+        // We will still be hanging on to anything in the range [-inf, 13).
+        // So we will have already returned 16, 14.
+        {15},
+        {1},
+    };
+
+    // Disable input order checking so we can see what happens.
+    sorter = makeDesc({}, /* checkInput */ false);
+    auto output = sort(input);
+    ASSERT_EQ(output.size(), 7);
+
+    ASSERT_EQ(output[0].time, 16);
+    ASSERT_EQ(output[1].time, 14);
+    ASSERT_EQ(output[2].time, 15);  // Out of order.
+    ASSERT_EQ(output[3].time, 10);
+    ASSERT_EQ(output[4].time, 5);
+    ASSERT_EQ(output[5].time, 3);
+    ASSERT_EQ(output[6].time, 1);
+
+    // Test that by default, bad input like this would be detected.
+    sorter = makeDesc({});
+    ASSERT(sorter->checkInput());
+    ASSERT_THROWS_CODE(sort(input), DBException, 6369910);
+}
 
 }  // namespace
 }  // namespace sorter
@@ -1311,5 +1427,9 @@ template class ::mongo::Sorter<::mongo::sorter::BoundedSorterTest::Key,
                                ::mongo::sorter::BoundedSorterTest::Doc>;
 template class ::mongo::BoundedSorter<::mongo::sorter::BoundedSorterTest::Key,
                                       ::mongo::sorter::BoundedSorterTest::Doc,
-                                      ::mongo::sorter::BoundedSorterTest::Comparator,
-                                      ::mongo::sorter::BoundedSorterTest::BoundMaker>;
+                                      ::mongo::sorter::BoundedSorterTest::ComparatorAsc,
+                                      ::mongo::sorter::BoundedSorterTest::BoundMakerAsc>;
+template class ::mongo::BoundedSorter<::mongo::sorter::BoundedSorterTest::Key,
+                                      ::mongo::sorter::BoundedSorterTest::Doc,
+                                      ::mongo::sorter::BoundedSorterTest::ComparatorDesc,
+                                      ::mongo::sorter::BoundedSorterTest::BoundMakerDesc>;