SERVER-25268 added StorageInterfaceImpl::findOne and deleteOne support for startKey and boundInclusion

6 files changed, 511 insertions, 113 deletions

diff --git a/src/mongo/db/repl/oplog_buffer_collection.cpp b/src/mongo/db/repl/oplog_buffer_collection.cpp
index 65082937bc8..e77431a54cd 100644
--- a/src/mongo/db/repl/oplog_buffer_collection.cpp
+++ b/src/mongo/db/repl/oplog_buffer_collection.cpp
@@ -217,7 +217,8 @@ bool OplogBufferCollection::_pop_inlock(OperationContext* txn, Value* value) {
         return true;
     }
     auto scanDirection = StorageInterface::ScanDirection::kForward;
-    auto result = _storageInterface->deleteOne(txn, _nss, kIdIdxName, scanDirection);
+    auto result = _storageInterface->deleteOne(
+        txn, _nss, kIdIdxName, scanDirection, {}, BoundInclusion::kIncludeStartKeyOnly);
     if (!result.isOK()) {
         if (result != ErrorCodes::CollectionIsEmpty) {
             fassert(40162, result.getStatus());
@@ -244,7 +245,8 @@ bool OplogBufferCollection::_peekOneSide_inlock(OperationContext* txn,
     }
     auto scanDirection = front ? StorageInterface::ScanDirection::kForward
                                : StorageInterface::ScanDirection::kBackward;
-    auto result = _storageInterface->findOne(txn, _nss, kIdIdxName, scanDirection);
+    auto result = _storageInterface->findOne(
+        txn, _nss, kIdIdxName, scanDirection, {}, BoundInclusion::kIncludeStartKeyOnly);
     if (!result.isOK()) {
         if (result != ErrorCodes::CollectionIsEmpty) {
             fassert(40163, result.getStatus());
diff --git a/src/mongo/db/repl/storage_interface.h b/src/mongo/db/repl/storage_interface.h
index bb72cfed19b..a65c7b2b0f3 100644
--- a/src/mongo/db/repl/storage_interface.h
+++ b/src/mongo/db/repl/storage_interface.h
@@ -36,6 +36,7 @@
 #include "mongo/base/disallow_copying.h"
 #include "mongo/base/string_data.h"
 #include "mongo/db/namespace_string.h"
+#include "mongo/db/query/index_bounds.h"
 #include "mongo/db/repl/collection_bulk_loader.h"
 #include "mongo/db/repl/optime.h"
 #include "mongo/db/service_context.h"
@@ -227,6 +228,12 @@ public:
      * Finds the first document returned by a collection or index scan on the collection in the
      * requested direction.
      * If "indexName" is boost::none, a collection scan is used to locate the document.
+     * Index scan options:
+     *     If "startKey" is not empty, the index scan will start from the given key (instead of
+     *     MinKey/MaxKey).
+     *     Set "boundInclusion" to BoundInclusion::kIncludeStartKeyOnly to include "startKey" in
+     *     the index scan results. Set to BoundInclusion::kIncludeEndKeyOnly to return the key
+     *     immediately following "startKey" from the index.
      */
     enum class ScanDirection {
         kForward = 1,
@@ -235,7 +242,9 @@ public:
     virtual StatusWith<BSONObj> findOne(OperationContext* txn,
                                         const NamespaceString& nss,
                                         boost::optional<StringData> indexName,
-                                        ScanDirection scanDirection) = 0;
+                                        ScanDirection scanDirection,
+                                        const BSONObj& startKey,
+                                        BoundInclusion boundInclusion) = 0;
 
     /**
      * Deletes the first document returned by a collection or index scan on the collection in the
@@ -245,7 +254,9 @@ public:
     virtual StatusWith<BSONObj> deleteOne(OperationContext* txn,
                                           const NamespaceString& nss,
                                           boost::optional<StringData> indexName,
-                                          ScanDirection scanDirection) = 0;
+                                          ScanDirection scanDirection,
+                                          const BSONObj& startKey,
+                                          BoundInclusion boundInclusion) = 0;
 };
 
 }  // namespace repl
diff --git a/src/mongo/db/repl/storage_interface_impl.cpp b/src/mongo/db/repl/storage_interface_impl.cpp
index c71191a2fce..7a568856493 100644
--- a/src/mongo/db/repl/storage_interface_impl.cpp
+++ b/src/mongo/db/repl/storage_interface_impl.cpp
@@ -457,6 +457,8 @@ StatusWith<BSONObj> _findOrDeleteOne(OperationContext* txn,
                                      const NamespaceString& nss,
                                      boost::optional<StringData> indexName,
                                      StorageInterface::ScanDirection scanDirection,
+                                     const BSONObj& startKey,
+                                     BoundInclusion boundInclusion,
                                      FindDeleteMode mode) {
     auto isFind = mode == FindDeleteMode::kFind;
     auto opStr = isFind ? "StorageInterfaceImpl::findOne" : "StorageInterfaceImpl::deleteOne";
@@ -476,6 +478,15 @@ StatusWith<BSONObj> _findOrDeleteOne(OperationContext* txn,
 
         std::unique_ptr<PlanExecutor> planExecutor;
         if (!indexName) {
+            if (!startKey.isEmpty()) {
+                return {ErrorCodes::NoSuchKey,
+                        "non-empty startKey not allowed for collection scan"};
+            }
+            if (boundInclusion != BoundInclusion::kIncludeStartKeyOnly) {
+                return {ErrorCodes::InvalidOptions,
+                        "bound inclusion must be BoundInclusion::kIncludeStartKeyOnly for "
+                        "collection scan"};
+            }
             // Use collection scan.
             planExecutor = isFind
                 ? InternalPlanner::collectionScan(
@@ -510,14 +521,16 @@ StatusWith<BSONObj> _findOrDeleteOne(OperationContext* txn,
             auto maxKey = Helpers::toKeyFormat(keyPattern.extendRangeBound({}, true));
             auto bounds =
                 isForward ? std::make_pair(minKey, maxKey) : std::make_pair(maxKey, minKey);
-
+            if (!startKey.isEmpty()) {
+                bounds.first = startKey;
+            }
             planExecutor = isFind
                 ? InternalPlanner::indexScan(txn,
                                              collection,
                                              indexDescriptor,
                                              bounds.first,
                                              bounds.second,
-                                             BoundInclusion::kIncludeStartKeyOnly,
+                                             boundInclusion,
                                              PlanExecutor::YIELD_MANUAL,
                                              direction,
                                              InternalPlanner::IXSCAN_FETCH)
@@ -527,7 +540,7 @@ StatusWith<BSONObj> _findOrDeleteOne(OperationContext* txn,
                                                        indexDescriptor,
                                                        bounds.first,
                                                        bounds.second,
-                                                       BoundInclusion::kIncludeStartKeyOnly,
+                                                       boundInclusion,
                                                        PlanExecutor::YIELD_MANUAL,
                                                        direction);
         }
@@ -550,15 +563,21 @@ StatusWith<BSONObj> _findOrDeleteOne(OperationContext* txn,
 StatusWith<BSONObj> StorageInterfaceImpl::findOne(OperationContext* txn,
                                                   const NamespaceString& nss,
                                                   boost::optional<StringData> indexName,
-                                                  ScanDirection scanDirection) {
-    return _findOrDeleteOne(txn, nss, indexName, scanDirection, FindDeleteMode::kFind);
+                                                  ScanDirection scanDirection,
+                                                  const BSONObj& startKey,
+                                                  BoundInclusion boundInclusion) {
+    return _findOrDeleteOne(
+        txn, nss, indexName, scanDirection, startKey, boundInclusion, FindDeleteMode::kFind);
 }
 
 StatusWith<BSONObj> StorageInterfaceImpl::deleteOne(OperationContext* txn,
                                                     const NamespaceString& nss,
                                                     boost::optional<StringData> indexName,
-                                                    ScanDirection scanDirection) {
-    return _findOrDeleteOne(txn, nss, indexName, scanDirection, FindDeleteMode::kDelete);
+                                                    ScanDirection scanDirection,
+                                                    const BSONObj& startKey,
+                                                    BoundInclusion boundInclusion) {
+    return _findOrDeleteOne(
+        txn, nss, indexName, scanDirection, startKey, boundInclusion, FindDeleteMode::kDelete);
 }
 
 Status StorageInterfaceImpl::isAdminDbValid(OperationContext* txn) {
diff --git a/src/mongo/db/repl/storage_interface_impl.h b/src/mongo/db/repl/storage_interface_impl.h
index 7586c81e062..8ddc2be0902 100644
--- a/src/mongo/db/repl/storage_interface_impl.h
+++ b/src/mongo/db/repl/storage_interface_impl.h
@@ -107,12 +107,16 @@ public:
     StatusWith<BSONObj> findOne(OperationContext* txn,
                                 const NamespaceString& nss,
                                 boost::optional<StringData> indexName,
-                                ScanDirection scanDirection) override;
+                                ScanDirection scanDirection,
+                                const BSONObj& startKey,
+                                BoundInclusion boundInclusion) override;
 
     StatusWith<BSONObj> deleteOne(OperationContext* txn,
                                   const NamespaceString& nss,
                                   boost::optional<StringData> indexName,
-                                  ScanDirection scanDirection) override;
+                                  ScanDirection scanDirection,
+                                  const BSONObj& startKey,
+                                  BoundInclusion boundInclusion) override;
 
     Status isAdminDbValid(OperationContext* txn) override;
 
diff --git a/src/mongo/db/repl/storage_interface_impl_test.cpp b/src/mongo/db/repl/storage_interface_impl_test.cpp
index 2ec1acfbdd6..ddac652c7ef 100644
--- a/src/mongo/db/repl/storage_interface_impl_test.cpp
+++ b/src/mongo/db/repl/storage_interface_impl_test.cpp
@@ -28,6 +28,7 @@
 
 #include "mongo/platform/basic.h"
 
+#include <algorithm>
 #include <boost/optional.hpp>
 #include <memory>
 
@@ -593,7 +594,12 @@ TEST_F(StorageInterfaceImplWithReplCoordTest, FindOneReturnsInvalidNamespaceIfCo
     auto nss = makeNamespace(_agent);
     auto indexName = "_id_"_sd;
     ASSERT_EQUALS(ErrorCodes::NamespaceNotFound,
-                  storage.findOne(txn, nss, indexName, StorageInterface::ScanDirection::kForward));
+                  storage.findOne(txn,
+                                  nss,
+                                  indexName,
+                                  StorageInterface::ScanDirection::kForward,
+                                  {},
+                                  BoundInclusion::kIncludeStartKeyOnly));
 }
 
 TEST_F(StorageInterfaceImplWithReplCoordTest, FindOneReturnsIndexNotFoundIfIndexIsMissing) {
@@ -603,7 +609,12 @@ TEST_F(StorageInterfaceImplWithReplCoordTest, FindOneReturnsIndexNotFoundIfIndex
     auto indexName = "nonexistent"_sd;
     ASSERT_OK(storage.createCollection(txn, nss, CollectionOptions()));
     ASSERT_EQUALS(ErrorCodes::IndexNotFound,
-                  storage.findOne(txn, nss, indexName, StorageInterface::ScanDirection::kForward));
+                  storage.findOne(txn,
+                                  nss,
+                                  indexName,
+                                  StorageInterface::ScanDirection::kForward,
+                                  {},
+                                  BoundInclusion::kIncludeStartKeyOnly));
 }
 
 TEST_F(StorageInterfaceImplWithReplCoordTest,
@@ -625,7 +636,12 @@ TEST_F(StorageInterfaceImplWithReplCoordTest,
     ASSERT_OK(loader->commit());
     auto indexName = "x_1"_sd;
     ASSERT_EQUALS(ErrorCodes::IndexOptionsConflict,
-                  storage.findOne(txn, nss, indexName, StorageInterface::ScanDirection::kForward));
+                  storage.findOne(txn,
+                                  nss,
+                                  indexName,
+                                  StorageInterface::ScanDirection::kForward,
+                                  {},
+                                  BoundInclusion::kIncludeStartKeyOnly));
 }
 
 TEST_F(StorageInterfaceImplWithReplCoordTest, FindOneReturnsCollectionIsEmptyIfCollectionIsEmpty) {
@@ -635,7 +651,28 @@ TEST_F(StorageInterfaceImplWithReplCoordTest, FindOneReturnsCollectionIsEmptyIfC
     auto indexName = "_id_"_sd;
     ASSERT_OK(storage.createCollection(txn, nss, CollectionOptions()));
     ASSERT_EQUALS(ErrorCodes::CollectionIsEmpty,
-                  storage.findOne(txn, nss, indexName, StorageInterface::ScanDirection::kForward));
+                  storage.findOne(txn,
+                                  nss,
+                                  indexName,
+                                  StorageInterface::ScanDirection::kForward,
+                                  {},
+                                  BoundInclusion::kIncludeStartKeyOnly));
+}
+
+/**
+ * Check collection contents. OplogInterface returns documents in reverse natural order.
+ */
+void _assertDocumentsInCollectionEquals(OperationContext* txn,
+                                        const NamespaceString& nss,
+                                        const std::vector<BSONObj>& docs) {
+    std::vector<BSONObj> reversedDocs(docs);
+    std::reverse(reversedDocs.begin(), reversedDocs.end());
+    OplogInterfaceLocal oplog(txn, nss.ns());
+    auto iter = oplog.makeIterator();
+    for (const auto& doc : reversedDocs) {
+        ASSERT_BSONOBJ_EQ(doc, unittest::assertGet(iter->next()).first);
+    }
+    ASSERT_EQUALS(ErrorCodes::CollectionIsEmpty, iter->next().getStatus());
 }
 
 TEST_F(StorageInterfaceImplWithReplCoordTest,
@@ -645,19 +682,87 @@ TEST_F(StorageInterfaceImplWithReplCoordTest,
     auto nss = makeNamespace(_agent);
     auto indexName = "_id_"_sd;
     ASSERT_OK(storage.createCollection(txn, nss, CollectionOptions()));
-    ASSERT_OK(
-        storage.insertDocuments(txn, nss, {BSON("_id" << 0), BSON("_id" << 1), BSON("_id" << 2)}));
+    ASSERT_OK(storage.insertDocuments(txn,
+                                      nss,
+                                      {BSON("_id" << 0),
+                                       BSON("_id" << 1),
+                                       BSON("_id" << 2),
+                                       BSON("_id" << 3),
+                                       BSON("_id" << 4)}));
+
+    // startKey not provided
+    ASSERT_BSONOBJ_EQ(BSON("_id" << 0),
+                      unittest::assertGet(storage.findOne(txn,
+                                                          nss,
+                                                          indexName,
+                                                          StorageInterface::ScanDirection::kForward,
+                                                          {},
+                                                          BoundInclusion::kIncludeStartKeyOnly)));
+
+    // startKey provided; include start key
     ASSERT_BSONOBJ_EQ(BSON("_id" << 0),
-                      unittest::assertGet(storage.findOne(
-                          txn, nss, indexName, StorageInterface::ScanDirection::kForward)));
+                      unittest::assertGet(storage.findOne(txn,
+                                                          nss,
+                                                          indexName,
+                                                          StorageInterface::ScanDirection::kForward,
+                                                          BSON("" << 0),
+                                                          BoundInclusion::kIncludeStartKeyOnly)));
+    ASSERT_BSONOBJ_EQ(BSON("_id" << 1),
+                      unittest::assertGet(storage.findOne(txn,
+                                                          nss,
+                                                          indexName,
+                                                          StorageInterface::ScanDirection::kForward,
+                                                          BSON("" << 1),
+                                                          BoundInclusion::kIncludeStartKeyOnly)));
 
-    // Check collection contents. OplogInterface returns documents in reverse natural order.
-    OplogInterfaceLocal oplog(txn, nss.ns());
-    auto iter = oplog.makeIterator();
-    ASSERT_BSONOBJ_EQ(BSON("_id" << 2), unittest::assertGet(iter->next()).first);
-    ASSERT_BSONOBJ_EQ(BSON("_id" << 1), unittest::assertGet(iter->next()).first);
-    ASSERT_BSONOBJ_EQ(BSON("_id" << 0), unittest::assertGet(iter->next()).first);
-    ASSERT_EQUALS(ErrorCodes::CollectionIsEmpty, iter->next().getStatus());
+    ASSERT_BSONOBJ_EQ(BSON("_id" << 1),
+                      unittest::assertGet(storage.findOne(txn,
+                                                          nss,
+                                                          indexName,
+                                                          StorageInterface::ScanDirection::kForward,
+                                                          BSON("" << 0.5),
+                                                          BoundInclusion::kIncludeStartKeyOnly)));
+
+    // startKey provided; include both start and end keys
+    ASSERT_BSONOBJ_EQ(BSON("_id" << 1),
+                      unittest::assertGet(storage.findOne(txn,
+                                                          nss,
+                                                          indexName,
+                                                          StorageInterface::ScanDirection::kForward,
+                                                          BSON("" << 1),
+                                                          BoundInclusion::kIncludeStartKeyOnly)));
+
+    // startKey provided; exclude start key
+    ASSERT_BSONOBJ_EQ(BSON("_id" << 2),
+                      unittest::assertGet(storage.findOne(txn,
+                                                          nss,
+                                                          indexName,
+                                                          StorageInterface::ScanDirection::kForward,
+                                                          BSON("" << 1),
+                                                          BoundInclusion::kIncludeEndKeyOnly)));
+
+    ASSERT_BSONOBJ_EQ(BSON("_id" << 2),
+                      unittest::assertGet(storage.findOne(txn,
+                                                          nss,
+                                                          indexName,
+                                                          StorageInterface::ScanDirection::kForward,
+                                                          BSON("" << 1.5),
+                                                          BoundInclusion::kIncludeEndKeyOnly)));
+
+    // startKey provided; exclude both start and end keys
+    ASSERT_BSONOBJ_EQ(
+        BSON("_id" << 2),
+        unittest::assertGet(storage.findOne(txn,
+                                            nss,
+                                            indexName,
+                                            StorageInterface::ScanDirection::kForward,
+                                            BSON("" << 1),
+                                            BoundInclusion::kExcludeBothStartAndEndKeys)));
+
+    _assertDocumentsInCollectionEquals(
+        txn,
+        nss,
+        {BSON("_id" << 0), BSON("_id" << 1), BSON("_id" << 2), BSON("_id" << 3), BSON("_id" << 4)});
 }
 
 TEST_F(StorageInterfaceImplWithReplCoordTest,
@@ -667,19 +772,76 @@ TEST_F(StorageInterfaceImplWithReplCoordTest,
     auto nss = makeNamespace(_agent);
     auto indexName = "_id_"_sd;
     ASSERT_OK(storage.createCollection(txn, nss, CollectionOptions()));
-    ASSERT_OK(
-        storage.insertDocuments(txn, nss, {BSON("_id" << 0), BSON("_id" << 1), BSON("_id" << 2)}));
-    ASSERT_BSONOBJ_EQ(BSON("_id" << 2),
-                      unittest::assertGet(storage.findOne(
-                          txn, nss, indexName, StorageInterface::ScanDirection::kBackward)));
-
-    // Check collection contents. OplogInterface returns documents in reverse natural order.
-    OplogInterfaceLocal oplog(txn, nss.ns());
-    auto iter = oplog.makeIterator();
-    ASSERT_BSONOBJ_EQ(BSON("_id" << 2), unittest::assertGet(iter->next()).first);
-    ASSERT_BSONOBJ_EQ(BSON("_id" << 1), unittest::assertGet(iter->next()).first);
-    ASSERT_BSONOBJ_EQ(BSON("_id" << 0), unittest::assertGet(iter->next()).first);
-    ASSERT_EQUALS(ErrorCodes::CollectionIsEmpty, iter->next().getStatus());
+    ASSERT_OK(storage.insertDocuments(txn,
+                                      nss,
+                                      {BSON("_id" << 0),
+                                       BSON("_id" << 1),
+                                       BSON("_id" << 2),
+                                       BSON("_id" << 3),
+                                       BSON("_id" << 4)}));
+
+    // startKey not provided
+    ASSERT_BSONOBJ_EQ(
+        BSON("_id" << 4),
+        unittest::assertGet(storage.findOne(txn,
+                                            nss,
+                                            indexName,
+                                            StorageInterface::ScanDirection::kBackward,
+                                            {},
+                                            BoundInclusion::kIncludeStartKeyOnly)));
+
+    // startKey provided; include start key
+    ASSERT_BSONOBJ_EQ(
+        BSON("_id" << 4),
+        unittest::assertGet(storage.findOne(txn,
+                                            nss,
+                                            indexName,
+                                            StorageInterface::ScanDirection::kBackward,
+                                            BSON("" << 4),
+                                            BoundInclusion::kIncludeStartKeyOnly)));
+    ASSERT_BSONOBJ_EQ(
+        BSON("_id" << 3),
+        unittest::assertGet(storage.findOne(txn,
+                                            nss,
+                                            indexName,
+                                            StorageInterface::ScanDirection::kBackward,
+                                            BSON("" << 3),
+                                            BoundInclusion::kIncludeStartKeyOnly)));
+
+    // startKey provided; include both start and end keys
+    ASSERT_BSONOBJ_EQ(
+        BSON("_id" << 4),
+        unittest::assertGet(storage.findOne(txn,
+                                            nss,
+                                            indexName,
+                                            StorageInterface::ScanDirection::kBackward,
+                                            BSON("" << 4),
+                                            BoundInclusion::kIncludeBothStartAndEndKeys)));
+
+    // startKey provided; exclude start key
+    ASSERT_BSONOBJ_EQ(
+        BSON("_id" << 2),
+        unittest::assertGet(storage.findOne(txn,
+                                            nss,
+                                            indexName,
+                                            StorageInterface::ScanDirection::kBackward,
+                                            BSON("" << 3),
+                                            BoundInclusion::kIncludeEndKeyOnly)));
+
+    // startKey provided; exclude both start and end keys
+    ASSERT_BSONOBJ_EQ(
+        BSON("_id" << 2),
+        unittest::assertGet(storage.findOne(txn,
+                                            nss,
+                                            indexName,
+                                            StorageInterface::ScanDirection::kBackward,
+                                            BSON("" << 3),
+                                            BoundInclusion::kExcludeBothStartAndEndKeys)));
+
+    _assertDocumentsInCollectionEquals(
+        txn,
+        nss,
+        {BSON("_id" << 0), BSON("_id" << 1), BSON("_id" << 2), BSON("_id" << 3), BSON("_id" << 4)});
 }
 
 TEST_F(StorageInterfaceImplWithReplCoordTest,
@@ -691,8 +853,12 @@ TEST_F(StorageInterfaceImplWithReplCoordTest,
     ASSERT_OK(
         storage.insertDocuments(txn, nss, {BSON("_id" << 1), BSON("_id" << 2), BSON("_id" << 0)}));
     ASSERT_BSONOBJ_EQ(BSON("_id" << 1),
-                      unittest::assertGet(storage.findOne(
-                          txn, nss, boost::none, StorageInterface::ScanDirection::kForward)));
+                      unittest::assertGet(storage.findOne(txn,
+                                                          nss,
+                                                          boost::none,
+                                                          StorageInterface::ScanDirection::kForward,
+                                                          {},
+                                                          BoundInclusion::kIncludeStartKeyOnly)));
 
     // Check collection contents. OplogInterface returns documents in reverse natural order.
     OplogInterfaceLocal oplog(txn, nss.ns());
@@ -711,17 +877,51 @@ TEST_F(StorageInterfaceImplWithReplCoordTest,
     ASSERT_OK(storage.createCollection(txn, nss, CollectionOptions()));
     ASSERT_OK(
         storage.insertDocuments(txn, nss, {BSON("_id" << 1), BSON("_id" << 2), BSON("_id" << 0)}));
-    ASSERT_BSONOBJ_EQ(BSON("_id" << 0),
-                      unittest::assertGet(storage.findOne(
-                          txn, nss, boost::none, StorageInterface::ScanDirection::kBackward)));
+    ASSERT_BSONOBJ_EQ(
+        BSON("_id" << 0),
+        unittest::assertGet(storage.findOne(txn,
+                                            nss,
+                                            boost::none,
+                                            StorageInterface::ScanDirection::kBackward,
+                                            {},
+                                            BoundInclusion::kIncludeStartKeyOnly)));
+
+    _assertDocumentsInCollectionEquals(
+        txn, nss, {BSON("_id" << 1), BSON("_id" << 2), BSON("_id" << 0)});
+}
 
-    // Check collection contents. OplogInterface returns documents in reverse natural order.
-    OplogInterfaceLocal oplog(txn, nss.ns());
-    auto iter = oplog.makeIterator();
-    ASSERT_BSONOBJ_EQ(BSON("_id" << 0), unittest::assertGet(iter->next()).first);
-    ASSERT_BSONOBJ_EQ(BSON("_id" << 2), unittest::assertGet(iter->next()).first);
-    ASSERT_BSONOBJ_EQ(BSON("_id" << 1), unittest::assertGet(iter->next()).first);
-    ASSERT_EQUALS(ErrorCodes::CollectionIsEmpty, iter->next().getStatus());
+TEST_F(StorageInterfaceImplWithReplCoordTest,
+       FindOneCollectionScanReturnsNoSuchKeyIfStartKeyIsNotEmpty) {
+    auto txn = getOperationContext();
+    StorageInterfaceImpl storage;
+    auto nss = makeNamespace(_agent);
+    ASSERT_OK(storage.createCollection(txn, nss, CollectionOptions()));
+    ASSERT_OK(
+        storage.insertDocuments(txn, nss, {BSON("_id" << 1), BSON("_id" << 2), BSON("_id" << 0)}));
+    ASSERT_EQUALS(ErrorCodes::NoSuchKey,
+                  storage.findOne(txn,
+                                  nss,
+                                  boost::none,
+                                  StorageInterface::ScanDirection::kForward,
+                                  BSON("" << 1),
+                                  BoundInclusion::kIncludeStartKeyOnly));
+}
+
+TEST_F(StorageInterfaceImplWithReplCoordTest,
+       FindOneCollectionScanReturnsInvalidOptionsIfBoundIsNotStartKeyOnly) {
+    auto txn = getOperationContext();
+    StorageInterfaceImpl storage;
+    auto nss = makeNamespace(_agent);
+    ASSERT_OK(storage.createCollection(txn, nss, CollectionOptions()));
+    ASSERT_OK(
+        storage.insertDocuments(txn, nss, {BSON("_id" << 1), BSON("_id" << 2), BSON("_id" << 0)}));
+    ASSERT_EQUALS(ErrorCodes::InvalidOptions,
+                  storage.findOne(txn,
+                                  nss,
+                                  boost::none,
+                                  StorageInterface::ScanDirection::kForward,
+                                  {},
+                                  BoundInclusion::kIncludeEndKeyOnly));
 }
 
 TEST_F(StorageInterfaceImplWithReplCoordTest,
@@ -730,9 +930,13 @@ TEST_F(StorageInterfaceImplWithReplCoordTest,
     StorageInterfaceImpl storage;
     auto nss = makeNamespace(_agent);
     auto indexName = "_id_"_sd;
-    ASSERT_EQUALS(
-        ErrorCodes::NamespaceNotFound,
-        storage.deleteOne(txn, nss, indexName, StorageInterface::ScanDirection::kForward));
+    ASSERT_EQUALS(ErrorCodes::NamespaceNotFound,
+                  storage.deleteOne(txn,
+                                    nss,
+                                    indexName,
+                                    StorageInterface::ScanDirection::kForward,
+                                    {},
+                                    BoundInclusion::kIncludeStartKeyOnly));
 }
 
 TEST_F(StorageInterfaceImplWithReplCoordTest, DeleteOneReturnsIndexNotFoundIfIndexIsMissing) {
@@ -741,9 +945,13 @@ TEST_F(StorageInterfaceImplWithReplCoordTest, DeleteOneReturnsIndexNotFoundIfInd
     auto nss = makeNamespace(_agent);
     auto indexName = "nonexistent"_sd;
     ASSERT_OK(storage.createCollection(txn, nss, CollectionOptions()));
-    ASSERT_EQUALS(
-        ErrorCodes::IndexNotFound,
-        storage.deleteOne(txn, nss, indexName, StorageInterface::ScanDirection::kForward));
+    ASSERT_EQUALS(ErrorCodes::IndexNotFound,
+                  storage.deleteOne(txn,
+                                    nss,
+                                    indexName,
+                                    StorageInterface::ScanDirection::kForward,
+                                    {},
+                                    BoundInclusion::kIncludeStartKeyOnly));
 }
 
 TEST_F(StorageInterfaceImplWithReplCoordTest,
@@ -753,9 +961,13 @@ TEST_F(StorageInterfaceImplWithReplCoordTest,
     auto nss = makeNamespace(_agent);
     auto indexName = "_id_"_sd;
     ASSERT_OK(storage.createCollection(txn, nss, CollectionOptions()));
-    ASSERT_EQUALS(
-        ErrorCodes::CollectionIsEmpty,
-        storage.deleteOne(txn, nss, indexName, StorageInterface::ScanDirection::kForward));
+    ASSERT_EQUALS(ErrorCodes::CollectionIsEmpty,
+                  storage.deleteOne(txn,
+                                    nss,
+                                    indexName,
+                                    StorageInterface::ScanDirection::kForward,
+                                    {},
+                                    BoundInclusion::kIncludeStartKeyOnly));
 }
 
 TEST_F(StorageInterfaceImplWithReplCoordTest,
@@ -765,18 +977,70 @@ TEST_F(StorageInterfaceImplWithReplCoordTest,
     auto nss = makeNamespace(_agent);
     auto indexName = "_id_"_sd;
     ASSERT_OK(storage.createCollection(txn, nss, CollectionOptions()));
-    ASSERT_OK(
-        storage.insertDocuments(txn, nss, {BSON("_id" << 0), BSON("_id" << 1), BSON("_id" << 2)}));
-    ASSERT_BSONOBJ_EQ(BSON("_id" << 0),
-                      unittest::assertGet(storage.deleteOne(
-                          txn, nss, indexName, StorageInterface::ScanDirection::kForward)));
-
-    // Check collection contents. OplogInterface returns documents in reverse natural order.
-    OplogInterfaceLocal oplog(txn, nss.ns());
-    auto iter = oplog.makeIterator();
-    ASSERT_BSONOBJ_EQ(BSON("_id" << 2), unittest::assertGet(iter->next()).first);
-    ASSERT_BSONOBJ_EQ(BSON("_id" << 1), unittest::assertGet(iter->next()).first);
-    ASSERT_EQUALS(ErrorCodes::CollectionIsEmpty, iter->next().getStatus());
+    ASSERT_OK(storage.insertDocuments(txn,
+                                      nss,
+                                      {BSON("_id" << 0),
+                                       BSON("_id" << 1),
+                                       BSON("_id" << 2),
+                                       BSON("_id" << 3),
+                                       BSON("_id" << 4),
+                                       BSON("_id" << 5),
+                                       BSON("_id" << 6),
+                                       BSON("_id" << 7)}));
+
+    // startKey not provided
+    ASSERT_BSONOBJ_EQ(
+        BSON("_id" << 0),
+        unittest::assertGet(storage.deleteOne(txn,
+                                              nss,
+                                              indexName,
+                                              StorageInterface::ScanDirection::kForward,
+                                              {},
+                                              BoundInclusion::kIncludeStartKeyOnly)));
+
+    _assertDocumentsInCollectionEquals(txn,
+                                       nss,
+                                       {BSON("_id" << 1),
+                                        BSON("_id" << 2),
+                                        BSON("_id" << 3),
+                                        BSON("_id" << 4),
+                                        BSON("_id" << 5),
+                                        BSON("_id" << 6),
+                                        BSON("_id" << 7)});
+
+    // startKey provided; include start key
+    ASSERT_BSONOBJ_EQ(
+        BSON("_id" << 2),
+        unittest::assertGet(storage.deleteOne(txn,
+                                              nss,
+                                              indexName,
+                                              StorageInterface::ScanDirection::kForward,
+                                              BSON("" << 2),
+                                              BoundInclusion::kIncludeStartKeyOnly)));
+
+    _assertDocumentsInCollectionEquals(txn,
+                                       nss,
+                                       {BSON("_id" << 1),
+                                        BSON("_id" << 3),
+                                        BSON("_id" << 4),
+                                        BSON("_id" << 5),
+                                        BSON("_id" << 6),
+                                        BSON("_id" << 7)});
+
+    // startKey provided; exclude start key
+    ASSERT_BSONOBJ_EQ(
+        BSON("_id" << 5),
+        unittest::assertGet(storage.deleteOne(txn,
+                                              nss,
+                                              indexName,
+                                              StorageInterface::ScanDirection::kForward,
+                                              BSON("" << 4),
+                                              BoundInclusion::kIncludeEndKeyOnly)));
+
+    _assertDocumentsInCollectionEquals(
+        txn,
+        nss,
+        {BSON("_id" << 1), BSON("_id" << 3), BSON("_id" << 4), BSON("_id" << 6), BSON("_id" << 7)});
 }
 
 TEST_F(StorageInterfaceImplWithReplCoordTest,
@@ -786,18 +1050,70 @@ TEST_F(StorageInterfaceImplWithReplCoordTest,
     auto nss = makeNamespace(_agent);
     auto indexName = "_id_"_sd;
     ASSERT_OK(storage.createCollection(txn, nss, CollectionOptions()));
-    ASSERT_OK(
-        storage.insertDocuments(txn, nss, {BSON("_id" << 0), BSON("_id" << 1), BSON("_id" << 2)}));
-    ASSERT_BSONOBJ_EQ(BSON("_id" << 2),
-                      unittest::assertGet(storage.deleteOne(
-                          txn, nss, indexName, StorageInterface::ScanDirection::kBackward)));
-
-    // Check collection contents. OplogInterface returns documents in reverse natural order.
-    OplogInterfaceLocal oplog(txn, nss.ns());
-    auto iter = oplog.makeIterator();
-    ASSERT_BSONOBJ_EQ(BSON("_id" << 1), unittest::assertGet(iter->next()).first);
-    ASSERT_BSONOBJ_EQ(BSON("_id" << 0), unittest::assertGet(iter->next()).first);
-    ASSERT_EQUALS(ErrorCodes::CollectionIsEmpty, iter->next().getStatus());
+    ASSERT_OK(storage.insertDocuments(txn,
+                                      nss,
+                                      {BSON("_id" << 0),
+                                       BSON("_id" << 1),
+                                       BSON("_id" << 2),
+                                       BSON("_id" << 3),
+                                       BSON("_id" << 4),
+                                       BSON("_id" << 5),
+                                       BSON("_id" << 6),
+                                       BSON("_id" << 7)}));
+
+    // startKey not provided
+    ASSERT_BSONOBJ_EQ(
+        BSON("_id" << 7),
+        unittest::assertGet(storage.deleteOne(txn,
+                                              nss,
+                                              indexName,
+                                              StorageInterface::ScanDirection::kBackward,
+                                              {},
+                                              BoundInclusion::kIncludeStartKeyOnly)));
+
+    _assertDocumentsInCollectionEquals(txn,
+                                       nss,
+                                       {BSON("_id" << 0),
+                                        BSON("_id" << 1),
+                                        BSON("_id" << 2),
+                                        BSON("_id" << 3),
+                                        BSON("_id" << 4),
+                                        BSON("_id" << 5),
+                                        BSON("_id" << 6)});
+
+    // startKey provided; include start key
+    ASSERT_BSONOBJ_EQ(
+        BSON("_id" << 5),
+        unittest::assertGet(storage.deleteOne(txn,
+                                              nss,
+                                              indexName,
+                                              StorageInterface::ScanDirection::kBackward,
+                                              BSON("" << 5),
+                                              BoundInclusion::kIncludeStartKeyOnly)));
+
+    _assertDocumentsInCollectionEquals(txn,
+                                       nss,
+                                       {BSON("_id" << 0),
+                                        BSON("_id" << 1),
+                                        BSON("_id" << 2),
+                                        BSON("_id" << 3),
+                                        BSON("_id" << 4),
+                                        BSON("_id" << 6)});
+
+    // startKey provided; exclude start key
+    ASSERT_BSONOBJ_EQ(
+        BSON("_id" << 2),
+        unittest::assertGet(storage.deleteOne(txn,
+                                              nss,
+                                              indexName,
+                                              StorageInterface::ScanDirection::kBackward,
+                                              BSON("" << 3),
+                                              BoundInclusion::kIncludeEndKeyOnly)));
+
+    _assertDocumentsInCollectionEquals(
+        txn,
+        nss,
+        {BSON("_id" << 0), BSON("_id" << 1), BSON("_id" << 3), BSON("_id" << 4), BSON("_id" << 6)});
 }
 
 TEST_F(StorageInterfaceImplWithReplCoordTest,
@@ -808,16 +1124,16 @@ TEST_F(StorageInterfaceImplWithReplCoordTest,
     ASSERT_OK(storage.createCollection(txn, nss, CollectionOptions()));
     ASSERT_OK(
         storage.insertDocuments(txn, nss, {BSON("_id" << 1), BSON("_id" << 2), BSON("_id" << 0)}));
-    ASSERT_BSONOBJ_EQ(BSON("_id" << 1),
-                      unittest::assertGet(storage.deleteOne(
-                          txn, nss, boost::none, StorageInterface::ScanDirection::kForward)));
-
-    // Check collection contents. OplogInterface returns documents in reverse natural order.
-    OplogInterfaceLocal oplog(txn, nss.ns());
-    auto iter = oplog.makeIterator();
-    ASSERT_BSONOBJ_EQ(BSON("_id" << 0), unittest::assertGet(iter->next()).first);
-    ASSERT_BSONOBJ_EQ(BSON("_id" << 2), unittest::assertGet(iter->next()).first);
-    ASSERT_EQUALS(ErrorCodes::CollectionIsEmpty, iter->next().getStatus());
+    ASSERT_BSONOBJ_EQ(
+        BSON("_id" << 1),
+        unittest::assertGet(storage.deleteOne(txn,
+                                              nss,
+                                              boost::none,
+                                              StorageInterface::ScanDirection::kForward,
+                                              {},
+                                              BoundInclusion::kIncludeStartKeyOnly)));
+
+    _assertDocumentsInCollectionEquals(txn, nss, {BSON("_id" << 2), BSON("_id" << 0)});
 }
 
 TEST_F(StorageInterfaceImplWithReplCoordTest,
@@ -828,16 +1144,50 @@ TEST_F(StorageInterfaceImplWithReplCoordTest,
     ASSERT_OK(storage.createCollection(txn, nss, CollectionOptions()));
     ASSERT_OK(
         storage.insertDocuments(txn, nss, {BSON("_id" << 1), BSON("_id" << 2), BSON("_id" << 0)}));
-    ASSERT_BSONOBJ_EQ(BSON("_id" << 0),
-                      unittest::assertGet(storage.deleteOne(
-                          txn, nss, boost::none, StorageInterface::ScanDirection::kBackward)));
+    ASSERT_BSONOBJ_EQ(
+        BSON("_id" << 0),
+        unittest::assertGet(storage.deleteOne(txn,
+                                              nss,
+                                              boost::none,
+                                              StorageInterface::ScanDirection::kBackward,
+                                              {},
+                                              BoundInclusion::kIncludeStartKeyOnly)));
+
+    _assertDocumentsInCollectionEquals(txn, nss, {BSON("_id" << 1), BSON("_id" << 2)});
+}
 
-    // Check collection contents. OplogInterface returns documents in reverse natural order.
-    OplogInterfaceLocal oplog(txn, nss.ns());
-    auto iter = oplog.makeIterator();
-    ASSERT_BSONOBJ_EQ(BSON("_id" << 2), unittest::assertGet(iter->next()).first);
-    ASSERT_BSONOBJ_EQ(BSON("_id" << 1), unittest::assertGet(iter->next()).first);
-    ASSERT_EQUALS(ErrorCodes::CollectionIsEmpty, iter->next().getStatus());
+TEST_F(StorageInterfaceImplWithReplCoordTest,
+       DeleteOneCollectionScanReturnsNoSuchKeyIfStartKeyIsNotEmpty) {
+    auto txn = getOperationContext();
+    StorageInterfaceImpl storage;
+    auto nss = makeNamespace(_agent);
+    ASSERT_OK(storage.createCollection(txn, nss, CollectionOptions()));
+    ASSERT_OK(
+        storage.insertDocuments(txn, nss, {BSON("_id" << 1), BSON("_id" << 2), BSON("_id" << 0)}));
+    ASSERT_EQUALS(ErrorCodes::NoSuchKey,
+                  storage.deleteOne(txn,
+                                    nss,
+                                    boost::none,
+                                    StorageInterface::ScanDirection::kForward,
+                                    BSON("" << 1),
+                                    BoundInclusion::kIncludeStartKeyOnly));
+}
+
+TEST_F(StorageInterfaceImplWithReplCoordTest,
+       DeleteOneCollectionScanReturnsInvalidOptionsIfBoundIsNotStartKeyOnly) {
+    auto txn = getOperationContext();
+    StorageInterfaceImpl storage;
+    auto nss = makeNamespace(_agent);
+    ASSERT_OK(storage.createCollection(txn, nss, CollectionOptions()));
+    ASSERT_OK(
+        storage.insertDocuments(txn, nss, {BSON("_id" << 1), BSON("_id" << 2), BSON("_id" << 0)}));
+    ASSERT_EQUALS(ErrorCodes::InvalidOptions,
+                  storage.deleteOne(txn,
+                                    nss,
+                                    boost::none,
+                                    StorageInterface::ScanDirection::kForward,
+                                    {},
+                                    BoundInclusion::kIncludeEndKeyOnly));
 }
 
 }  // namespace
diff --git a/src/mongo/db/repl/storage_interface_mock.h b/src/mongo/db/repl/storage_interface_mock.h
index f2b680b18f1..6eb5bf14556 100644
--- a/src/mongo/db/repl/storage_interface_mock.h
+++ b/src/mongo/db/repl/storage_interface_mock.h
@@ -109,11 +109,15 @@ public:
     using FindOneFn = stdx::function<StatusWith<BSONObj>(OperationContext* txn,
                                                          const NamespaceString& nss,
                                                          boost::optional<StringData> indexName,
-                                                         ScanDirection scanDirection)>;
+                                                         ScanDirection scanDirection,
+                                                         const BSONObj& startKey,
+                                                         BoundInclusion boundInclusion)>;
     using DeleteOneFn = stdx::function<StatusWith<BSONObj>(OperationContext* txn,
                                                            const NamespaceString& nss,
                                                            boost::optional<StringData> indexName,
-                                                           ScanDirection scanDirection)>;
+                                                           ScanDirection scanDirection,
+                                                           const BSONObj& startKey,
+                                                           BoundInclusion boundInclusion)>;
     using IsAdminDbValidFn = stdx::function<Status(OperationContext* txn)>;
 
     StorageInterfaceMock() = default;
@@ -178,15 +182,19 @@ public:
     StatusWith<BSONObj> findOne(OperationContext* txn,
                                 const NamespaceString& nss,
                                 boost::optional<StringData> indexName,
-                                ScanDirection scanDirection) override {
-        return findOneFn(txn, nss, indexName, scanDirection);
+                                ScanDirection scanDirection,
+                                const BSONObj& startKey,
+                                BoundInclusion boundInclusion) override {
+        return findOneFn(txn, nss, indexName, scanDirection, startKey, boundInclusion);
     }
 
     StatusWith<BSONObj> deleteOne(OperationContext* txn,
                                   const NamespaceString& nss,
                                   boost::optional<StringData> indexName,
-                                  ScanDirection scanDirection) override {
-        return deleteOneFn(txn, nss, indexName, scanDirection);
+                                  ScanDirection scanDirection,
+                                  const BSONObj& startKey,
+                                  BoundInclusion boundInclusion) override {
+        return deleteOneFn(txn, nss, indexName, scanDirection, startKey, boundInclusion);
     }
 
     Status isAdminDbValid(OperationContext* txn) override {
@@ -227,13 +235,17 @@ public:
     FindOneFn findOneFn = [](OperationContext* txn,
                              const NamespaceString& nss,
                              boost::optional<StringData> indexName,
-                             ScanDirection scanDirection) {
+                             ScanDirection scanDirection,
+                             const BSONObj& startKey,
+                             BoundInclusion boundInclusion) {
         return Status{ErrorCodes::IllegalOperation, "FindOneFn not implemented."};
     };
     DeleteOneFn deleteOneFn = [](OperationContext* txn,
                                  const NamespaceString& nss,
                                  boost::optional<StringData> indexName,
-                                 ScanDirection scanDirection) {
+                                 ScanDirection scanDirection,
+                                 const BSONObj& startKey,
+                                 BoundInclusion boundInclusion) {
         return Status{ErrorCodes::IllegalOperation, "DeleteOneFn not implemented."};
     };
     IsAdminDbValidFn isAdminDbValidFn = [](OperationContext*) {