SERVER-42972 Callers of SortedDataInterface::seekExact should pass KeyString

12 files changed, 237 insertions, 170 deletions

diff --git a/src/mongo/db/index/index_access_method.cpp b/src/mongo/db/index/index_access_method.cpp
index 989ce31802e..c8cfae7aa88 100644
--- a/src/mongo/db/index/index_access_method.cpp
+++ b/src/mongo/db/index/index_access_method.cpp
@@ -239,11 +239,7 @@ Status AbstractIndexAccessMethod::touch(OperationContext* opCtx, const BSONObj&
 
     std::unique_ptr<SortedDataInterface::Cursor> cursor(_newInterface->newCursor(opCtx));
     for (const auto& keyString : keys) {
-        auto key = KeyString::toBson(keyString.getBuffer(),
-                                     keyString.getSize(),
-                                     getSortedDataInterface()->getOrdering(),
-                                     keyString.getTypeBits());
-        cursor->seekExact(key);
+        cursor->seekExact(keyString);
     }
 
     return Status::OK();
@@ -257,37 +253,40 @@ Status AbstractIndexAccessMethod::touch(OperationContext* opCtx) const {
 RecordId AbstractIndexAccessMethod::findSingle(OperationContext* opCtx,
                                                const BSONObj& requestedKey) const {
     // Generate the key for this index.
-    boost::optional<KeyString::Value> actualKey;
-    if (_btreeState->getCollator()) {
-        // For performance, call get keys only if there is a non-simple collation.
-        KeyStringSet keys;
-        KeyStringSet* multikeyMetadataKeys = nullptr;
-        MultikeyPaths* multikeyPaths = nullptr;
-        getKeys(requestedKey,
-                GetKeysMode::kEnforceConstraints,
-                &keys,
-                multikeyMetadataKeys,
-                multikeyPaths);
-        invariant(keys.size() == 1);
-        actualKey.emplace(std::move(*keys.begin()));
-    } else {
-        KeyString::HeapBuilder requestedKeyString(getSortedDataInterface()->getKeyStringVersion(),
-                                                  BSONObj::stripFieldNames(requestedKey),
-                                                  getSortedDataInterface()->getOrdering());
-        actualKey.emplace(requestedKeyString.release());
-    }
+    KeyString::Value actualKey = [&]() {
+        if (_btreeState->getCollator()) {
+            // For performance, call get keys only if there is a non-simple collation.
+            KeyStringSet keys;
+            KeyStringSet* multikeyMetadataKeys = nullptr;
+            MultikeyPaths* multikeyPaths = nullptr;
+            getKeys(requestedKey,
+                    GetKeysMode::kEnforceConstraints,
+                    &keys,
+                    multikeyMetadataKeys,
+                    multikeyPaths);
+            invariant(keys.size() == 1);
+            return *keys.begin();
+        } else {
+            KeyString::HeapBuilder requestedKeyString(
+                getSortedDataInterface()->getKeyStringVersion(),
+                BSONObj::stripFieldNames(requestedKey),
+                getSortedDataInterface()->getOrdering());
+            return requestedKeyString.release();
+        }
+    }();
 
     std::unique_ptr<SortedDataInterface::Cursor> cursor(_newInterface->newCursor(opCtx));
     const auto requestedInfo = kDebugBuild ? SortedDataInterface::Cursor::kKeyAndLoc
                                            : SortedDataInterface::Cursor::kWantLoc;
-    auto key = KeyString::toBson(actualKey->getBuffer(),
-                                 actualKey->getSize(),
-                                 getSortedDataInterface()->getOrdering(),
-                                 actualKey->getTypeBits());
-    if (auto kv = cursor->seekExact(key, requestedInfo)) {
+    if (auto kv = cursor->seekExact(actualKey, requestedInfo)) {
         // StorageEngine should guarantee these.
         dassert(!kv->loc.isNull());
-        dassert(kv->key.woCompare(key, /*order*/ BSONObj(), /*considerFieldNames*/ false) == 0);
+        dassert(kv->key.woCompare(KeyString::toBson(actualKey.getBuffer(),
+                                                    actualKey.getSize(),
+                                                    getSortedDataInterface()->getOrdering(),
+                                                    actualKey.getTypeBits()),
+                                  /*order*/ BSONObj(),
+                                  /*considerFieldNames*/ false) == 0);
 
         return kv->loc;
     }
diff --git a/src/mongo/db/storage/biggie/biggie_sorted_impl.cpp b/src/mongo/db/storage/biggie/biggie_sorted_impl.cpp
index ac627f6d5c0..05166bc7307 100644
--- a/src/mongo/db/storage/biggie/biggie_sorted_impl.cpp
+++ b/src/mongo/db/storage/biggie/biggie_sorted_impl.cpp
@@ -848,26 +848,13 @@ boost::optional<KeyStringEntry> SortedDataInterface::Cursor::seekForKeyString(
     return seekAfterProcessing(keyStringValue);
 }
 
-boost::optional<IndexKeyEntry> SortedDataInterface::Cursor::seekExact(const BSONObj& key,
-                                                                      RequestedInfo) {
-    BSONObj finalKey = BSONObj::stripFieldNames(key);
-    KeyString::Builder keyString(KeyString::Version::V1, finalKey, _order);
-    auto ksEntry = seekExact(keyString.getValueCopy());
-    if (ksEntry) {
-        const BSONObj bson = KeyString::toBson(ksEntry->keyString.getBuffer(),
-                                               ksEntry->keyString.getSize(),
-                                               _order,
-                                               ksEntry->keyString.getTypeBits());
-        auto kv = seekAfterProcessing(bson);
-        if (kv) {
-            return kv;
-        }
-    }
-    return {};
-}
-
-boost::optional<KeyStringEntry> SortedDataInterface::Cursor::seekExact(
+boost::optional<KeyStringEntry> SortedDataInterface::Cursor::seekExactForKeyString(
     const KeyString::Value& keyStringValue) {
+    dassert(KeyString::decodeDiscriminator(keyStringValue.getBuffer(),
+                                           keyStringValue.getSize(),
+                                           _order,
+                                           keyStringValue.getTypeBits()) ==
+            KeyString::Discriminator::kInclusive);
     auto ksEntry = seekForKeyString(keyStringValue);
     if (!ksEntry) {
         return {};
@@ -882,6 +869,23 @@ boost::optional<KeyStringEntry> SortedDataInterface::Cursor::seekExact(
     return {};
 }
 
+boost::optional<IndexKeyEntry> SortedDataInterface::Cursor::seekExact(
+    const KeyString::Value& keyStringValue, RequestedInfo parts) {
+    auto ksEntry = seekExactForKeyString(keyStringValue);
+    if (!ksEntry) {
+        return {};
+    }
+
+    BSONObj bson;
+    if (parts & SortedDataInterface::Cursor::kWantKey) {
+        bson = KeyString::toBson(ksEntry->keyString.getBuffer(),
+                                 ksEntry->keyString.getSize(),
+                                 _order,
+                                 ksEntry->keyString.getTypeBits());
+    }
+    return IndexKeyEntry(std::move(bson), ksEntry->loc);
+}
+
 void SortedDataInterface::Cursor::save() {
     _atEOF = false;
     if (_lastMoveWasRestore) {
diff --git a/src/mongo/db/storage/biggie/biggie_sorted_impl.h b/src/mongo/db/storage/biggie/biggie_sorted_impl.h
index eef51fdd53a..c2c6f638fd2 100644
--- a/src/mongo/db/storage/biggie/biggie_sorted_impl.h
+++ b/src/mongo/db/storage/biggie/biggie_sorted_impl.h
@@ -122,10 +122,10 @@ public:
                                                     RequestedInfo parts = kKeyAndLoc) override;
         virtual boost::optional<KeyStringEntry> seekForKeyString(
             const KeyString::Value& keyStringValue) override;
-        virtual boost::optional<IndexKeyEntry> seekExact(const BSONObj& key,
-                                                         RequestedInfo parts = kKeyAndLoc) override;
-        virtual boost::optional<KeyStringEntry> seekExact(
+        virtual boost::optional<KeyStringEntry> seekExactForKeyString(
             const KeyString::Value& keyStringValue) override;
+        virtual boost::optional<IndexKeyEntry> seekExact(const KeyString::Value& keyStringValue,
+                                                         RequestedInfo) override;
         virtual void save() override;
         virtual void restore() override;
         virtual void detachFromOperationContext() override;
diff --git a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_btree_impl.cpp b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_btree_impl.cpp
index 3bc907f6721..7053db41a66 100644
--- a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_btree_impl.cpp
+++ b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_btree_impl.cpp
@@ -357,21 +357,26 @@ public:
             }
         }
 
-        boost::optional<IndexKeyEntry> seekExact(const BSONObj& key, RequestedInfo) {
-            auto kv = _seek(key, true, kKeyAndLoc);
-            if (kv && kv->key.woCompare(key, BSONObj(), /*considerFieldNames*/ false) == 0)
+        boost::optional<IndexKeyEntry> _seekExact(const BSONObj& key, RequestedInfo parts) {
+            auto kv = _seek(key, true, parts);
+            if (!kv || kv->key.woCompare(key, BSONObj(), /*considerFieldNames*/ false) != 0)
+                return {};
+
+            if (parts & SortedDataInterface::Cursor::kWantKey) {
                 return kv;
-            return {};
+            }
+            return IndexKeyEntry{{}, kv->loc};
         }
 
-        boost::optional<KeyStringEntry> seekExact(const KeyString::Value& keyStringValue) {
+        boost::optional<KeyStringEntry> seekExactForKeyString(
+            const KeyString::Value& keyStringValue) override {
             const BSONObj query = KeyString::toBson(keyStringValue.getBuffer(),
                                                     keyStringValue.getSize(),
                                                     _ordering,
                                                     keyStringValue.getTypeBits());
-            auto kv = seekExact(query, kKeyAndLoc);
+            auto kv = _seekExact(query, kKeyAndLoc);
             if (kv) {
-                // We have retrived a valid result from seekExact(). Convert to KeyString
+                // We have retrived a valid result from _seekExact(). Convert to KeyString
                 // and return
                 KeyString::Builder ks(KeyString::Version::V1, kv->key, _ordering);
                 ks.appendRecordId(kv->loc);
@@ -380,6 +385,15 @@ public:
             return {};
         }
 
+        boost::optional<IndexKeyEntry> seekExact(const KeyString::Value& keyStringValue,
+                                                 RequestedInfo parts) override {
+            const BSONObj query = KeyString::toBson(keyStringValue.getBuffer(),
+                                                    keyStringValue.getSize(),
+                                                    _ordering,
+                                                    keyStringValue.getTypeBits());
+            return _seekExact(query, parts);
+        }
+
         void save() override {
             // Keep original position if we haven't moved since the last restore.
             _opCtx = nullptr;
diff --git a/src/mongo/db/storage/key_string.cpp b/src/mongo/db/storage/key_string.cpp
index b55eeb3f4e5..764bb78f04a 100644
--- a/src/mongo/db/storage/key_string.cpp
+++ b/src/mongo/db/storage/key_string.cpp
@@ -398,7 +398,7 @@ void BuilderBase<BufferT>::appendSetAsArray(const BSONElementSet& set, const Str
 }
 
 template <class BufferT>
-void BuilderBase<BufferT>::_appendDiscriminator(const Discriminator discriminator) {
+void BuilderBase<BufferT>::appendDiscriminator(const Discriminator discriminator) {
     // The discriminator forces this KeyString to compare Less/Greater than any KeyString with
     // the same prefix of keys. As an example, this can be used to land on the first key in the
     // index with the value "a" regardless of the RecordId. In compound indexes it can use a
@@ -414,8 +414,8 @@ void BuilderBase<BufferT>::_appendDiscriminator(const Discriminator discriminato
             break;  // No discriminator byte.
     }
 
-    // TODO consider omitting kEnd when using a discriminator byte. It is not a storage format
-    // change since keystrings with discriminators are not allowed to be stored.
+    // TODO (SERVER-43178): consider omitting kEnd when using a discriminator byte. It is not a
+    // storage format change since keystrings with discriminators are not allowed to be stored.
     _appendEnd();
 }
 // ----------------------------------------------------------------------
@@ -449,7 +449,7 @@ void BuilderBase<BufferT>::_appendAllElementsForIndexing(const BSONObj& obj,
             invariant(!it.more());
         }
     }
-    _appendDiscriminator(discriminator);
+    appendDiscriminator(discriminator);
 }
 
 template <class BufferT>
@@ -2451,13 +2451,11 @@ BSONObj toBsonSafeWithDiscriminator(const char* buffer,
 }
 
 // This discriminator byte only exists in KeyStrings for queries, not in KeyStrings stored in an
-// index. This function is only used by Biggie because it needs to extract the discriminator to do
-// the query.
+// index.
 Discriminator decodeDiscriminator(const char* buffer,
                                   size_t len,
                                   Ordering ord,
                                   const TypeBits& typeBits) {
-    BSONObjBuilder builder;
     BufReader reader(buffer, len);
     TypeBits::Reader typeBitsReader(typeBits);
     for (int i = 0; reader.remaining(); i++) {
@@ -2473,7 +2471,7 @@ Discriminator decodeDiscriminator(const char* buffer,
 
         if (ctype == kEnd)
             break;
-        toBsonValue(ctype, &reader, &typeBitsReader, invert, typeBits.version, &(builder << ""), 1);
+        filterKeyFromKeyString(ctype, &reader, invert, typeBits.version);
     }
     return Discriminator::kInclusive;
 }
diff --git a/src/mongo/db/storage/key_string.h b/src/mongo/db/storage/key_string.h
index 74fd5fd4d91..d4e0aebb24b 100644
--- a/src/mongo/db/storage/key_string.h
+++ b/src/mongo/db/storage/key_string.h
@@ -557,6 +557,11 @@ public:
     void appendSetAsArray(const BSONElementSet& set, const StringTransformFn& f = nullptr);
 
     /**
+     * Appends a Discriminator byte and kEnd byte to a key string.
+     */
+    void appendDiscriminator(const Discriminator discriminator);
+
+    /**
      * Resets to an empty state.
      * Equivalent to but faster than *this = Builder(ord, discriminator)
      */
@@ -664,7 +669,6 @@ private:
     void _appendDoubleWithoutTypeBits(const double num, DecimalContinuationMarker dcm, bool invert);
     void _appendHugeDecimalWithoutTypeBits(const Decimal128 dec, bool invert);
     void _appendTinyDecimalWithoutTypeBits(const Decimal128 dec, const double bin, bool invert);
-    void _appendDiscriminator(const Discriminator discriminator);
     void _appendEnd();
 
     template <typename T>
@@ -676,7 +680,7 @@ private:
 
     void _doneAppending() {
         if (_state == BuildState::kAppendingBSONElements) {
-            _appendDiscriminator(_discriminator);
+            appendDiscriminator(_discriminator);
         }
     }
 
@@ -697,7 +701,7 @@ private:
 
         switch (_state) {
             case BuildState::kEmpty:
-                invariant(to == BuildState::kAppendingBSONElements ||
+                invariant(to == BuildState::kAppendingBSONElements || to == BuildState::kEndAdded ||
                           to == BuildState::kAppendedRecordID);
                 break;
             case BuildState::kAppendingBSONElements:
diff --git a/src/mongo/db/storage/mobile/mobile_index.cpp b/src/mongo/db/storage/mobile/mobile_index.cpp
index 7a6281195a1..7851a290bab 100644
--- a/src/mongo/db/storage/mobile/mobile_index.cpp
+++ b/src/mongo/db/storage/mobile/mobile_index.cpp
@@ -355,7 +355,6 @@ public:
           _opCtx(opCtx),
           _isForward(isForward),
           _savedKey(index.getKeyStringVersion()),
-          _savedKeyWithoutDiscriminator(index.getKeyStringVersion()),
           _savedRecId(0),
           _savedTypeBits(index.getKeyStringVersion()),
           _startPosition(index.getKeyStringVersion()) {
@@ -432,51 +431,41 @@ public:
         return KeyStringEntry(_savedKey.getValueCopy(), _savedRecId);
     }
 
-    boost::optional<IndexKeyEntry> seekExact(const BSONObj& key, RequestedInfo) override {
-        // Create a separate KeyString that doesn't have a discriminator so that we can
-        // do a comparison with the retrieved KeyString.
-        if (!_isForward) {
-            _savedKeyWithoutDiscriminator.resetToKey(BSONObj::stripFieldNames(key),
-                                                     _index.getOrdering(),
-                                                     KeyString::Discriminator::kInclusive);
-        }
-        // If it's a reverse cursor, a kExclusiveAfter discriminator is included to ensure that
-        // the KeyString we construct will always be greater than the KeyString that we retrieve
-        // (as it might have a RecordId). So if our cursor lands on the exact match,
-        // it does not advance to the next key (in the reverse direction)
-        const auto discriminator = _isForward ? KeyString::Discriminator::kInclusive
-                                              : KeyString::Discriminator::kExclusiveAfter;
-        KeyString::Builder keyString(_index.getKeyStringVersion(),
-                                     BSONObj::stripFieldNames(key),
-                                     _index.getOrdering(),
-                                     discriminator);
-        auto ksEntry = seekExact(keyString.getValueCopy());
+    boost::optional<IndexKeyEntry> seekExact(const KeyString::Value& keyStringValue,
+                                             RequestedInfo parts) override {
+        auto ksEntry = seekExactForKeyString(keyStringValue);
         if (ksEntry) {
-            auto kv = getCurrentEntry(kKeyAndLoc);
+            auto kv = getCurrentEntry(parts);
             invariant(kv);
             return kv;
         }
         return {};
     }
 
-    boost::optional<KeyStringEntry> seekExact(const KeyString::Value& keyStringValue) override {
-        auto ksEntry = seekForKeyString(keyStringValue);
-        if (!ksEntry) {
-            return {};
-        }
-        // If it's a reverse cursor, we compare the KeyString we retrieved with the KeyString that
-        // doesn't have a discriminator.
-        if (!_isForward) {
-            if (KeyString::compare(
-                    ksEntry->keyString.getBuffer(),
-                    _savedKeyWithoutDiscriminator.getBuffer(),
-                    KeyString::sizeWithoutRecordIdAtEnd(ksEntry->keyString.getBuffer(),
-                                                        ksEntry->keyString.getSize()),
-                    _savedKeyWithoutDiscriminator.getSize()) == 0) {
-                return KeyStringEntry(ksEntry->keyString, ksEntry->loc);
+    boost::optional<KeyStringEntry> seekExactForKeyString(
+        const KeyString::Value& keyStringValue) override {
+        auto ksEntry = [&]() {
+            if (_isForward) {
+                return seekForKeyString(keyStringValue);
             }
+
+            // Append a kExclusiveAfter discriminator if it's a reverse cursor to ensure that the
+            // KeyString we construct will always be greater than the KeyString that we retrieve
+            // (even when it has a RecordId).
+            KeyString::Builder keyCopy(_index.getKeyStringVersion(), _index.getOrdering());
+
+            // Reset by copying all but the last byte, the kEnd byte.
+            keyCopy.resetFromBuffer(keyStringValue.getBuffer(), keyStringValue.getSize() - 1);
+
+            // Append a different discriminator and new end byte.
+            keyCopy.appendDiscriminator(KeyString::Discriminator::kExclusiveAfter);
+            return seekForKeyString(keyCopy.getValueCopy());
+        }();
+
+        if (!ksEntry) {
             return {};
         }
+
         if (KeyString::compare(ksEntry->keyString.getBuffer(),
                                keyStringValue.getBuffer(),
                                KeyString::sizeWithoutRecordIdAtEnd(ksEntry->keyString.getBuffer(),
@@ -631,7 +620,6 @@ protected:
     bool _isEOF = true;
 
     KeyString::Builder _savedKey;
-    KeyString::Builder _savedKeyWithoutDiscriminator;
     RecordId _savedRecId;
     KeyString::TypeBits _savedTypeBits;
 
diff --git a/src/mongo/db/storage/sorted_data_interface.h b/src/mongo/db/storage/sorted_data_interface.h
index 024fef895c7..b52d5f52374 100644
--- a/src/mongo/db/storage/sorted_data_interface.h
+++ b/src/mongo/db/storage/sorted_data_interface.h
@@ -288,11 +288,26 @@ public:
          * Seeks to a key with a hint to the implementation that you only want exact matches. If
          * an exact match can't be found, boost::none will be returned and the resulting
          * position of the cursor is unspecified.
+         *
+         * This will not accept a KeyString with a Discriminator other than kInclusive. Since
+         * keys are not stored with Discriminators, an exact match would never be found.
          */
-        virtual boost::optional<IndexKeyEntry> seekExact(const BSONObj& key,
-                                                         RequestedInfo parts = kKeyAndLoc) = 0;
+        virtual boost::optional<KeyStringEntry> seekExactForKeyString(
+            const KeyString::Value& keyString) = 0;
 
-        virtual boost::optional<KeyStringEntry> seekExact(const KeyString::Value& keyString) = 0;
+        /**
+         * Seeks to a key with a hint to the implementation that you only want exact matches. If
+         * an exact match can't be found, boost::none will be returned and the resulting
+         * position of the cursor is unspecified.
+         *
+         * This will not accept a KeyString with a Discriminator other than kInclusive. Since
+         * keys are not stored with Discriminators, an exact match would never be found.
+         *
+         * Unlike the previous method, this one will return IndexKeyEntry if an exact match is
+         * found.
+         */
+        virtual boost::optional<IndexKeyEntry> seekExact(const KeyString::Value& keyString,
+                                                         RequestedInfo parts = kKeyAndLoc) = 0;
 
         //
         // Saving and restoring state
diff --git a/src/mongo/db/storage/sorted_data_interface_test_cursor_end_position.cpp b/src/mongo/db/storage/sorted_data_interface_test_cursor_end_position.cpp
index 129657e4912..af88b08d13e 100644
--- a/src/mongo/db/storage/sorted_data_interface_test_cursor_end_position.cpp
+++ b/src/mongo/db/storage/sorted_data_interface_test_cursor_end_position.cpp
@@ -142,25 +142,25 @@ void testSetEndPosition_Seek_Forward(bool unique, bool inclusive) {
     cursor->setEndPosition(key3, inclusive);
 
     // Directly seeking past end is considered out of range.
-    ASSERT_EQ(cursor->seek(makeKeyStringForSeek(sorted.get(), key4, true, true)), boost::none);
-    ASSERT_EQ(cursor->seekExact(key4), boost::none);
+    ASSERT_EQ(cursor->seek(makeKeyStringForSeek(sorted.get(), key4, true, inclusive)), boost::none);
+    ASSERT_EQ(cursor->seekExact(makeKeyString(sorted.get(), key4)), boost::none);
 
     // Seeking to key3 directly or indirectly is only returned if endPosition is inclusive.
     auto maybeKey3 = inclusive ? boost::make_optional(IndexKeyEntry(key3, loc1)) : boost::none;
 
     // direct
-    ASSERT_EQ(cursor->seek(makeKeyStringForSeek(sorted.get(), key3, true, true)), maybeKey3);
-    ASSERT_EQ(cursor->seekExact(key3), maybeKey3);
+    ASSERT_EQ(cursor->seek(makeKeyStringForSeek(sorted.get(), key3, true, inclusive)), maybeKey3);
+    ASSERT_EQ(cursor->seekExact(makeKeyString(sorted.get(), key3)), maybeKey3);
 
     // indirect
-    ASSERT_EQ(cursor->seek(makeKeyStringForSeek(sorted.get(), key2, true, true)), maybeKey3);
+    ASSERT_EQ(cursor->seek(makeKeyStringForSeek(sorted.get(), key2, true, inclusive)), maybeKey3);
 
     cursor->saveUnpositioned();
     removeFromIndex(opCtx, sorted, {{key3, loc1}});
     cursor->restore();
 
-    ASSERT_EQ(cursor->seek(makeKeyStringForSeek(sorted.get(), key2, true, true)), boost::none);
-    ASSERT_EQ(cursor->seek(makeKeyStringForSeek(sorted.get(), key3, true, true)), boost::none);
+    ASSERT_EQ(cursor->seek(makeKeyStringForSeek(sorted.get(), key2, true, inclusive)), boost::none);
+    ASSERT_EQ(cursor->seek(makeKeyStringForSeek(sorted.get(), key3, true, inclusive)), boost::none);
 }
 TEST(SortedDataInterface, SetEndPosition_Seek_Forward_Unique_Inclusive) {
     testSetEndPosition_Seek_Forward(true, true);
@@ -191,15 +191,16 @@ void testSetEndPosition_Seek_Reverse(bool unique, bool inclusive) {
     cursor->setEndPosition(key2, inclusive);
 
     // Directly seeking past end is considered out of range.
-    ASSERT_EQ(cursor->seek(makeKeyStringForSeek(sorted.get(), key1, false, true)), boost::none);
-    ASSERT_EQ(cursor->seekExact(key1), boost::none);
+    ASSERT_EQ(cursor->seek(makeKeyStringForSeek(sorted.get(), key1, false, inclusive)),
+              boost::none);
+    ASSERT_EQ(cursor->seekExact(makeKeyString(sorted.get(), key1)), boost::none);
 
     // Seeking to key2 directly or indirectly is only returned if endPosition is inclusive.
     auto maybeKey2 = inclusive ? boost::make_optional(IndexKeyEntry(key2, loc1)) : boost::none;
 
     // direct
-    ASSERT_EQ(cursor->seek(makeKeyStringForSeek(sorted.get(), key2, false, true)), maybeKey2);
-    ASSERT_EQ(cursor->seekExact(key2), maybeKey2);
+    ASSERT_EQ(cursor->seek(makeKeyStringForSeek(sorted.get(), key2, false, inclusive)), maybeKey2);
+    ASSERT_EQ(cursor->seekExact(makeKeyString(sorted.get(), key2)), maybeKey2);
 
     // indirect
     ASSERT_EQ(cursor->seek(makeKeyStringForSeek(sorted.get(), key3, false, true)), maybeKey2);
diff --git a/src/mongo/db/storage/sorted_data_interface_test_cursor_seek_exact.cpp b/src/mongo/db/storage/sorted_data_interface_test_cursor_seek_exact.cpp
index 4a0584e0559..af88bd9215c 100644
--- a/src/mongo/db/storage/sorted_data_interface_test_cursor_seek_exact.cpp
+++ b/src/mongo/db/storage/sorted_data_interface_test_cursor_seek_exact.cpp
@@ -50,7 +50,12 @@ void testSeekExact_Hit(bool unique, bool forward) {
 
     auto cursor = sorted->newCursor(opCtx.get(), forward);
 
-    ASSERT_EQ(cursor->seekExact(key2), IndexKeyEntry(key2, loc1));
+    ASSERT_EQ(cursor->seekExact(makeKeyString(sorted.get(), key2)), IndexKeyEntry(key2, loc1));
+
+    // Only return the requested RecordId.
+    ASSERT_EQ(
+        cursor->seekExact(makeKeyString(sorted.get(), key2), SortedDataInterface::Cursor::kWantLoc),
+        IndexKeyEntry({}, loc1));
 
     // Make sure iterating works. We may consider loosening this requirement if it is a hardship
     // for some storage engines.
@@ -84,11 +89,11 @@ void testSeekExact_Miss(bool unique, bool forward) {
 
     auto cursor = sorted->newCursor(opCtx.get(), forward);
 
-    ASSERT_EQ(cursor->seekExact(key2), boost::none);
+    ASSERT_EQ(cursor->seekExact(makeKeyString(sorted.get(), key2)), boost::none);
 
     // Not testing iteration since the cursors position following a failed seekExact is
     // undefined. However, you must be able to seek somewhere else.
-    ASSERT_EQ(cursor->seekExact(key1), IndexKeyEntry(key1, loc1));
+    ASSERT_EQ(cursor->seekExact(makeKeyString(sorted.get(), key1)), IndexKeyEntry(key1, loc1));
 }
 TEST(SortedDataInterface, SeekExact_Miss_Unique_Forward) {
     testSeekExact_Miss(true, true);
@@ -120,7 +125,7 @@ TEST(SortedDataInterface, SeekExact_HitWithDups_Forward) {
 
     auto cursor = sorted->newCursor(opCtx.get());
 
-    ASSERT_EQ(cursor->seekExact(key2), IndexKeyEntry(key2, loc1));
+    ASSERT_EQ(cursor->seekExact(makeKeyString(sorted.get(), key2)), IndexKeyEntry(key2, loc1));
     ASSERT_EQ(cursor->next(), IndexKeyEntry(key2, loc2));
     ASSERT_EQ(cursor->next(), IndexKeyEntry(key3, loc1));
     ASSERT_EQ(cursor->next(), boost::none);
@@ -143,7 +148,7 @@ TEST(SortedDataInterface, SeekExact_HitWithDups_Reverse) {
 
     auto cursor = sorted->newCursor(opCtx.get(), false);
 
-    ASSERT_EQ(cursor->seekExact(key2), IndexKeyEntry(key2, loc2));
+    ASSERT_EQ(cursor->seekExact(makeKeyString(sorted.get(), key2)), IndexKeyEntry(key2, loc2));
     ASSERT_EQ(cursor->next(), IndexKeyEntry(key2, loc1));
     ASSERT_EQ(cursor->next(), IndexKeyEntry(key1, loc1));
     ASSERT_EQ(cursor->next(), boost::none);
diff --git a/src/mongo/db/storage/sorted_data_interface_test_insert.cpp b/src/mongo/db/storage/sorted_data_interface_test_insert.cpp
index 8ee6f4732db..74de3c0ec3e 100644
--- a/src/mongo/db/storage/sorted_data_interface_test_insert.cpp
+++ b/src/mongo/db/storage/sorted_data_interface_test_insert.cpp
@@ -595,11 +595,57 @@ TEST(SortedDataInterface, InsertAndSeekExactKeyString) {
 
         const std::unique_ptr<SortedDataInterface::Cursor> cursor(sorted->newCursor(opCtx.get()));
 
-        auto ksEntry1 = cursor->seekExact(keyString1WithoutRecordId);
+        auto ksEntry1 = cursor->seekExactForKeyString(keyString1WithoutRecordId);
         ASSERT_EQUALS(ksEntry1->keyString.compare(keyString1), 0);
         ASSERT_EQUALS(ksEntry1->keyString.compare(keyString2), -1);
 
-        auto ksEntry2 = cursor->seekExact(keyString2WithoutRecordId);
+        auto ksEntry2 = cursor->seekExactForKeyString(keyString2WithoutRecordId);
+        ASSERT_EQUALS(ksEntry2->keyString.compare(keyString2), 0);
+        ASSERT_EQUALS(ksEntry2->keyString.compare(keyString1), 1);
+    }
+}
+
+/*
+ * Insert multiple KeyStrings and seekExact to the inserted KeyStrings with reverse cursor.
+ */
+TEST(SortedDataInterface, InsertAndSeekExactKeyString_Reverse) {
+    const auto harnessHelper(newSortedDataInterfaceHarnessHelper());
+    const std::unique_ptr<SortedDataInterface> sorted(
+        harnessHelper->newSortedDataInterface(/*unique=*/true, /*partial=*/false));
+
+    auto keyString1 = makeKeyString(sorted.get(), key1, loc1);
+    auto keyString2 = makeKeyString(sorted.get(), key2, loc2);
+
+    auto keyString1WithoutRecordId = makeKeyString(sorted.get(), key1);
+    auto keyString2WithoutRecordId = makeKeyString(sorted.get(), key2);
+
+    {
+        const ServiceContext::UniqueOperationContext opCtx(harnessHelper->newOperationContext());
+        ASSERT(sorted->isEmpty(opCtx.get()));
+    }
+
+    {
+        const ServiceContext::UniqueOperationContext opCtx(harnessHelper->newOperationContext());
+        {
+            WriteUnitOfWork uow(opCtx.get());
+            ASSERT_OK(sorted->insert(opCtx.get(), keyString1, false));
+            ASSERT_OK(sorted->insert(opCtx.get(), keyString2, false));
+            uow.commit();
+        }
+    }
+
+    {
+        const ServiceContext::UniqueOperationContext opCtx(harnessHelper->newOperationContext());
+        ASSERT_EQUALS(2, sorted->numEntries(opCtx.get()));
+
+        const std::unique_ptr<SortedDataInterface::Cursor> cursor(
+            sorted->newCursor(opCtx.get(), false));
+
+        auto ksEntry1 = cursor->seekExactForKeyString(keyString1WithoutRecordId);
+        ASSERT_EQUALS(ksEntry1->keyString.compare(keyString1), 0);
+        ASSERT_EQUALS(ksEntry1->keyString.compare(keyString2), -1);
+
+        auto ksEntry2 = cursor->seekExactForKeyString(keyString2WithoutRecordId);
         ASSERT_EQUALS(ksEntry2->keyString.compare(keyString2), 0);
         ASSERT_EQUALS(ksEntry2->keyString.compare(keyString1), 1);
     }
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp
index 079ef58d6f1..b911c67f590 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp
@@ -801,7 +801,6 @@ public:
           _key(idx.getKeyStringVersion()),
           _typeBits(idx.getKeyStringVersion()),
           _query(idx.getKeyStringVersion()),
-          _queryWithoutDiscriminator(idx.getKeyStringVersion()),
           _prefix(prefix) {
         _cursor.emplace(_idx.uri(), _idx.tableId(), false, _opCtx);
     }
@@ -867,59 +866,54 @@ public:
         return KeyStringEntry(_key.getValueCopy(), _id);
     }
 
-    boost::optional<IndexKeyEntry> seekExact(const BSONObj& key, RequestedInfo) override {
-        // Create a separate KeyString that doesn't have a discriminator so that
-        // we can do a comparison with the retrieved KeyString.
-        if (!_forward) {
-            _queryWithoutDiscriminator.resetToKey(BSONObj::stripFieldNames(key),
-                                                  _idx.getOrdering(),
-                                                  KeyString::Discriminator::kInclusive);
-        }
-        // If it's a reverse cursor, a kExclusiveAfter discriminator is included to ensure that
-        // the KeyString we construct will always be greater than the KeyString that we retrieve
-        // (as it might have a RecordId). So if our cursor lands on the exact match,
-        // it does not advance to the next key (in the reverse direction)
-        const auto discriminator = _forward ? KeyString::Discriminator::kInclusive
-                                            : KeyString::Discriminator::kExclusiveAfter;
-        _query.resetToKey(BSONObj::stripFieldNames(key), _idx.getOrdering(), discriminator);
-        auto ksEntry = seekExact(_query.getValueCopy());
-        if (ksEntry) {
-            auto kv = curr(kKeyAndLoc);
-            invariant(kv);
-            return kv;
-        }
-        return {};
-    }
+    boost::optional<KeyStringEntry> seekExactForKeyString(const KeyString::Value& key) override {
+        dassert(KeyString::decodeDiscriminator(
+                    key.getBuffer(), key.getSize(), _idx.getOrdering(), key.getTypeBits()) ==
+                KeyString::Discriminator::kInclusive);
+
+        auto ksEntry = [&]() {
+            if (_forward) {
+                return seekForKeyString(key);
+            }
+
+            // Append a kExclusiveAfter discriminator if it's a reverse cursor to ensure that the
+            // KeyString we construct will always be greater than the KeyString that we retrieve
+            // (even when it has a RecordId).
+            KeyString::Builder keyCopy(_idx.getKeyStringVersion(), _idx.getOrdering());
+
+            // Reset by copying all but the last byte, the kEnd byte.
+            keyCopy.resetFromBuffer(key.getBuffer(), key.getSize() - 1);
+
+            // Append a different discriminator and new end byte.
+            keyCopy.appendDiscriminator(KeyString::Discriminator::kExclusiveAfter);
+            return seekForKeyString(keyCopy.getValueCopy());
+        }();
 
-    boost::optional<KeyStringEntry> seekExact(const KeyString::Value& keyStringValue) override {
-        auto ksEntry = seekForKeyString(keyStringValue);
         if (!ksEntry) {
             return {};
         }
 
-        // If it's a reverse cursor, we compare the KeyString we retrieved with the KeyString that
-        // doesn't have a discriminator.
-        if (!_forward) {
-            if (KeyString::compare(
-                    ksEntry->keyString.getBuffer(),
-                    _queryWithoutDiscriminator.getBuffer(),
-                    KeyString::sizeWithoutRecordIdAtEnd(ksEntry->keyString.getBuffer(),
-                                                        ksEntry->keyString.getSize()),
-                    _queryWithoutDiscriminator.getSize()) == 0) {
-                return KeyStringEntry(ksEntry->keyString, ksEntry->loc);
-            }
-            return {};
-        }
         if (KeyString::compare(ksEntry->keyString.getBuffer(),
-                               keyStringValue.getBuffer(),
+                               key.getBuffer(),
                                KeyString::sizeWithoutRecordIdAtEnd(ksEntry->keyString.getBuffer(),
                                                                    ksEntry->keyString.getSize()),
-                               keyStringValue.getSize()) == 0) {
+                               key.getSize()) == 0) {
             return KeyStringEntry(ksEntry->keyString, ksEntry->loc);
         }
         return {};
     }
 
+    boost::optional<IndexKeyEntry> seekExact(const KeyString::Value& keyStringValue,
+                                             RequestedInfo parts) override {
+        auto ksEntry = seekExactForKeyString(keyStringValue);
+        if (ksEntry) {
+            auto kv = curr(parts);
+            invariant(kv);
+            return kv;
+        }
+        return {};
+    }
+
     void save() override {
         try {
             if (_cursor)
@@ -1183,7 +1177,6 @@ protected:
     bool _lastMoveSkippedKey = false;
 
     KeyString::Builder _query;
-    KeyString::Builder _queryWithoutDiscriminator;
     KVPrefix _prefix;
 
     std::unique_ptr<KeyString::Builder> _endPosition;