SERVER-32756 Enable selection of V2 format unique index

A new unique index format would get added in MongoDB4.0 to overcome the
anonmaly of duplicate keys in unique index on secondary instances. Enabled
selection of new V2 format unique index via a gating variable. The
IndexDescriptor::IndexVersion now has a new flag kV2Unique. The version in an
IndexDescriptor object can be either kV2 or kV2Unique. A version value kV2Unique
means a V2 format unique index would be created. In this commit all of the
V2 format unique index implemetation is a copy of V1(unique index) format. It
would change in future commits.

10 files changed, 444 insertions, 7 deletions

diff --git a/jstests/noPassthrough/index_version_v2.js b/jstests/noPassthrough/index_version_v2.js
index e84f206394a..c6cc4b6d500 100644
--- a/jstests/noPassthrough/index_version_v2.js
+++ b/jstests/noPassthrough/index_version_v2.js
@@ -4,6 +4,9 @@
  * Additionally, this file tests that index version v=2 is required to create an index with a
  * collation and that index version v=2 is required to index decimal data on storage engines using
  * the KeyString format.
+ *
+ * Also, index version v=3 is required to prohibit duplicates in unique index at secondary. Enhance
+ * the tests for index version v=3.
  */
 (function() {
     "use strict";
@@ -118,6 +121,16 @@
 
     testDB.dropDatabase();
 
-    // Test that attempting to create an index with v=3 returns an error.
-    assert.commandFailed(testDB.index_version.createIndex({withV3: 1}, {v: 3}));
+    // MongoDB4.0 onwards index version v=3 would be supported. Test that an index created with v=3
+    // succeeds.
+    assert.commandWorked(testDB.index_version.createIndex({withV3: 1}, {v: 3}));
+
+    //
+    // Index version v=4
+    //
+
+    testDB.dropDatabase();
+
+    // Test that attempting to create an index with v=4 returns an error.
+    assert.commandFailed(testDB.index_version.createIndex({withV4: 1}, {v: 4}));
 })();
diff --git a/src/mongo/db/catalog/index_key_validate.cpp b/src/mongo/db/catalog/index_key_validate.cpp
index ca5ac2aaea5..4c1695069d5 100644
--- a/src/mongo/db/catalog/index_key_validate.cpp
+++ b/src/mongo/db/catalog/index_key_validate.cpp
@@ -133,7 +133,8 @@ Status validateKeyPattern(const BSONObj& key, IndexDescriptor::IndexVersion inde
 
                 break;
             }
-            case IndexVersion::kV2: {
+            case IndexVersion::kV2:
+            case IndexVersion::kV2Unique: {
                 if (keyElement.isNumber()) {
                     double value = keyElement.number();
                     if (std::isnan(value)) {
diff --git a/src/mongo/db/catalog/index_spec_validate_test.cpp b/src/mongo/db/catalog/index_spec_validate_test.cpp
index f0107cff856..471f1a525ca 100644
--- a/src/mongo/db/catalog/index_spec_validate_test.cpp
+++ b/src/mongo/db/catalog/index_spec_validate_test.cpp
@@ -299,7 +299,7 @@ TEST(IndexSpecValidateTest, ReturnsAnErrorIfVersionIsUnsupported) {
                                 BSON("key" << BSON("field" << 1) << "name"
                                            << "indexName"
                                            << "v"
-                                           << 3
+                                           << 4
                                            << "collation"
                                            << BSON("locale"
                                                    << "en")),
@@ -398,6 +398,29 @@ TEST(IndexSpecValidateTest, AcceptsIndexVersionV1) {
                       sorted(result.getValue()));
 }
 
+TEST(IndexSpecValidateTest, AcceptsIndexVersionV2Unique) {
+    ServerGlobalParams::FeatureCompatibility featureCompatibility;
+    featureCompatibility.setVersion(
+        ServerGlobalParams::FeatureCompatibility::Version::kFullyUpgradedTo36);
+
+    auto result = validateIndexSpec(kDefaultOpCtx,
+                                    BSON("key" << BSON("field" << 1) << "name"
+                                               << "indexName"
+                                               << "v"
+                                               << 3),
+                                    kTestNamespace,
+                                    featureCompatibility);
+    ASSERT_OK(result.getStatus());
+
+    // We don't care about the order of the fields in the resulting index specification.
+    ASSERT_BSONOBJ_EQ(sorted(BSON("key" << BSON("field" << 1) << "name"
+                                        << "indexName"
+                                        << "ns"
+                                        << kTestNamespace.ns()
+                                        << "v"
+                                        << 3)),
+                      sorted(result.getValue()));
+}
 TEST(IndexSpecValidateTest, ReturnsAnErrorIfCollationIsNotAnObject) {
     ASSERT_EQ(ErrorCodes::TypeMismatch,
               validateIndexSpec(kDefaultOpCtx,
diff --git a/src/mongo/db/index/btree_key_generator.cpp b/src/mongo/db/index/btree_key_generator.cpp
index d1737ddf314..9e24016e3e7 100644
--- a/src/mongo/db/index/btree_key_generator.cpp
+++ b/src/mongo/db/index/btree_key_generator.cpp
@@ -77,6 +77,7 @@ std::unique_ptr<BtreeKeyGenerator> BtreeKeyGenerator::make(IndexVersion indexVer
             return stdx::make_unique<BtreeKeyGeneratorV0>(fieldNames, fixed, isSparse);
         case IndexVersion::kV1:
         case IndexVersion::kV2:
+        case IndexVersion::kV2Unique:
             return stdx::make_unique<BtreeKeyGeneratorV1>(fieldNames, fixed, isSparse, collator);
     }
     return nullptr;
diff --git a/src/mongo/db/index/index_descriptor.cpp b/src/mongo/db/index/index_descriptor.cpp
index df9f2044563..9c72069f191 100644
--- a/src/mongo/db/index/index_descriptor.cpp
+++ b/src/mongo/db/index/index_descriptor.cpp
@@ -101,6 +101,7 @@ bool IndexDescriptor::isIndexVersionSupported(IndexVersion indexVersion) {
         case IndexVersion::kV0:
         case IndexVersion::kV1:
         case IndexVersion::kV2:
+        case IndexVersion::kV2Unique:
             return true;
     }
     return false;
@@ -119,6 +120,7 @@ Status IndexDescriptor::isIndexVersionAllowedForCreation(
             break;
         case IndexVersion::kV1:
         case IndexVersion::kV2:
+        case IndexVersion::kV2Unique:
             return Status::OK();
     }
     return {ErrorCodes::CannotCreateIndex,
@@ -129,6 +131,11 @@ Status IndexDescriptor::isIndexVersionAllowedForCreation(
 
 IndexVersion IndexDescriptor::getDefaultIndexVersion(
     ServerGlobalParams::FeatureCompatibility::Version featureCompatibilityVersion) {
+    // The gating variable would allow creation of V2 format unique index when set to true.
+    const bool useV2UniqueIndexFormat = false;
+    if (useV2UniqueIndexFormat)
+        return IndexVersion::kV2Unique;
+
     return IndexVersion::kV2;
 }
 
diff --git a/src/mongo/db/index/index_descriptor.h b/src/mongo/db/index/index_descriptor.h
index 3fae7128915..f5c8a46b76c 100644
--- a/src/mongo/db/index/index_descriptor.h
+++ b/src/mongo/db/index/index_descriptor.h
@@ -56,7 +56,7 @@ class OperationContext;
  */
 class IndexDescriptor {
 public:
-    enum class IndexVersion { kV0 = 0, kV1 = 1, kV2 = 2 };
+    enum class IndexVersion { kV0 = 0, kV1 = 1, kV2 = 2, kV2Unique = 3 };
     static constexpr IndexVersion kLatestIndexVersion = IndexVersion::kV2;
 
     static constexpr StringData k2dIndexBitsFieldName = "bits"_sd;
diff --git a/src/mongo/db/storage/mmap_v1/btree/btree_interface.cpp b/src/mongo/db/storage/mmap_v1/btree/btree_interface.cpp
index 14a3e57503b..11943d5479e 100644
--- a/src/mongo/db/storage/mmap_v1/btree/btree_interface.cpp
+++ b/src/mongo/db/storage/mmap_v1/btree/btree_interface.cpp
@@ -428,6 +428,7 @@ SortedDataInterface* getMMAPV1Interface(HeadManager* headManager,
                 headManager, recordStore, cursorRegistry, ordering, indexName, isUnique);
         case IndexVersion::kV1:
         case IndexVersion::kV2:
+        case IndexVersion::kV2Unique:
             return new BtreeInterfaceImpl<BtreeLayoutV1>(
                 headManager, recordStore, cursorRegistry, ordering, indexName, isUnique);
     }
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp
index a65ce5eff8a..7f0adbd589e 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp
@@ -396,6 +396,11 @@ Status WiredTigerIndex::dupKeyCheck(OperationContext* opCtx,
     invariant(!hasFieldNames(key));
     invariant(unique());
 
+    if (isV2FormatUniqueIndex()) {
+        // This is stubbed; specific implementation would come later.
+        return Status::OK();
+    }
+
     WiredTigerCursor curwrap(_uri, _tableId, false, opCtx);
     WT_CURSOR* c = curwrap.get();
 
@@ -474,6 +479,12 @@ long long WiredTigerIndex::getSpaceUsedBytes(OperationContext* opCtx) const {
 
 bool WiredTigerIndex::isDup(WT_CURSOR* c, const BSONObj& key, const RecordId& id) {
     invariant(unique());
+
+    if (isV2FormatUniqueIndex()) {
+        // This is stubbed; specific implementation would come later.
+        return false;
+    }
+
     // First check whether the key exists.
     KeyString data(keyStringVersion(), key, _ordering);
     WiredTigerItem item(data.getBuffer(), data.getSize());
@@ -710,6 +721,101 @@ private:
     std::vector<std::pair<RecordId, KeyString::TypeBits>> _records;
 };
 
+/**
+ * Bulk builds a V2 format unique index.
+ *
+ * This is a copy of `UniqueBulkBuilder` at the moment; it would change later.
+ *
+ * In order to support V2 format unique indexes in dupsAllowed mode this class only does an actual
+ * insert after it sees a key after the one we are trying to insert. This allows us to gather up all
+ * duplicate ids and insert them all together. This is necessary since bulk cursors can only
+ * append data.
+ */
+class WiredTigerIndex::UniqueV2BulkBuilder : public BulkBuilder {
+public:
+    UniqueV2BulkBuilder(WiredTigerIndex* idx,
+                        OperationContext* opCtx,
+                        bool dupsAllowed,
+                        KVPrefix prefix)
+        : BulkBuilder(idx, opCtx, prefix),
+          _idx(idx),
+          _dupsAllowed(dupsAllowed),
+          _keyString(idx->keyStringVersion()) {}
+
+    Status addKey(const BSONObj& newKey, const RecordId& id) {
+        {
+            const Status s = checkKeySize(newKey);
+            if (!s.isOK())
+                return s;
+        }
+
+        const int cmp = newKey.woCompare(_key, _ordering);
+        if (cmp != 0) {
+            if (!_key.isEmpty()) {   // _key.isEmpty() is only true on the first call to addKey().
+                invariant(cmp > 0);  // newKey must be > the last key
+                // We are done with dups of the last key so we can insert it now.
+                doInsert();
+            }
+            invariant(_records.empty());
+        } else {
+            // Dup found!
+            if (!_dupsAllowed) {
+                return _idx->dupKeyError(newKey);
+            }
+
+            // If we get here, we are in the weird mode where dups are allowed on a unique
+            // index, so add ourselves to the list of duplicate ids. This also replaces the
+            // _key which is correct since any dups seen later are likely to be newer.
+        }
+
+        _key = newKey.getOwned();
+        _keyString.resetToKey(_key, _idx->ordering());
+        _records.push_back(std::make_pair(id, _keyString.getTypeBits()));
+
+        return Status::OK();
+    }
+
+    void commit(bool mayInterrupt) {
+        WriteUnitOfWork uow(_opCtx);
+        if (!_records.empty()) {
+            // This handles inserting the last unique key.
+            doInsert();
+        }
+        uow.commit();
+    }
+
+private:
+    void doInsert() {
+        invariant(!_records.empty());
+
+        KeyString value(_idx->keyStringVersion());
+        for (size_t i = 0; i < _records.size(); i++) {
+            value.appendRecordId(_records[i].first);
+            // When there is only one record, we can omit AllZeros TypeBits. Otherwise they need
+            // to be included.
+            if (!(_records[i].second.isAllZeros() && _records.size() == 1)) {
+                value.appendTypeBits(_records[i].second);
+            }
+        }
+
+        WiredTigerItem keyItem(_keyString.getBuffer(), _keyString.getSize());
+        WiredTigerItem valueItem(value.getBuffer(), value.getSize());
+
+        setKey(_cursor, keyItem.Get());
+        _cursor->set_value(_cursor, valueItem.Get());
+
+        invariantWTOK(_cursor->insert(_cursor));
+
+        _records.clear();
+    }
+
+    WiredTigerIndex* _idx;
+    const bool _dupsAllowed;
+    BSONObj _key;
+    KeyString _keyString;
+    std::vector<std::pair<RecordId, KeyString::TypeBits>> _records;
+};
+
 namespace {
 
 /**
@@ -1102,6 +1208,54 @@ public:
     }
 };
 
+/**
+ * This is duplicate of `WiredTigerIndexUniqueCursor` at the moment; it would change later.
+ */
+class WiredTigerIndexUniqueV2Cursor final : public WiredTigerIndexCursorBase {
+public:
+    WiredTigerIndexUniqueV2Cursor(const WiredTigerIndex& idx,
+                                  OperationContext* opCtx,
+                                  bool forward,
+                                  KVPrefix prefix)
+        : WiredTigerIndexCursorBase(idx, opCtx, forward, prefix) {}
+
+    void updateIdAndTypeBits() override {
+        // We assume that cursors can only ever see unique indexes in their "pristine" state,
+        // where no duplicates are possible. The cases where dups are allowed should hold
+        // sufficient locks to ensure that no cursor ever sees them.
+        WT_CURSOR* c = _cursor->get();
+        WT_ITEM item;
+        invariantWTOK(c->get_value(c, &item));
+
+        BufReader br(item.data, item.size);
+        _id = KeyString::decodeRecordId(&br);
+        _typeBits.resetFromBuffer(&br);
+
+        if (!br.atEof()) {
+            severe() << "Unique index cursor seeing multiple records for key "
+                     << redact(curr(kWantKey)->key) << " in index " << _idx.indexName();
+            fassertFailed(40686);
+        }
+    }
+
+    boost::optional<IndexKeyEntry> seekExact(const BSONObj& key, RequestedInfo parts) override {
+        _query.resetToKey(stripFieldNames(key), _idx.ordering());
+        const WiredTigerItem keyItem(_query.getBuffer(), _query.getSize());
+
+        WT_CURSOR* c = _cursor->get();
+        setKey(c, keyItem.Get());
+
+        // Using search rather than search_near.
+        int ret = WT_READ_CHECK(c->search(c));
+        if (ret != WT_NOTFOUND)
+            invariantWTOK(ret);
+        _cursorAtEof = ret == WT_NOTFOUND;
+        updatePosition();
+        dassert(_eof || _key.compare(_query) == 0);
+        return curr(parts);
+    }
+};
+
 }  // namespace
 
 WiredTigerIndexUnique::WiredTigerIndexUnique(OperationContext* ctx,
@@ -1116,11 +1270,31 @@ std::unique_ptr<SortedDataInterface::Cursor> WiredTigerIndexUnique::newCursor(
     return stdx::make_unique<WiredTigerIndexUniqueCursor>(*this, opCtx, forward, _prefix);
 }
 
+/**
+ * This is duplicate of `WiredTigerIndexUnique` at the moment; might change later.
+ */
+WiredTigerIndexUniqueV2::WiredTigerIndexUniqueV2(OperationContext* ctx,
+                                                 const std::string& uri,
+                                                 const IndexDescriptor* desc,
+                                                 KVPrefix prefix,
+                                                 bool isReadOnly)
+    : WiredTigerIndex(ctx, uri, desc, prefix, isReadOnly), _partial(desc->isPartial()) {}
+
+std::unique_ptr<SortedDataInterface::Cursor> WiredTigerIndexUniqueV2::newCursor(
+    OperationContext* opCtx, bool forward) const {
+    return stdx::make_unique<WiredTigerIndexUniqueV2Cursor>(*this, opCtx, forward, _prefix);
+}
+
 SortedDataBuilderInterface* WiredTigerIndexUnique::getBulkBuilder(OperationContext* opCtx,
                                                                   bool dupsAllowed) {
     return new UniqueBulkBuilder(this, opCtx, dupsAllowed, _prefix);
 }
 
+SortedDataBuilderInterface* WiredTigerIndexUniqueV2::getBulkBuilder(OperationContext* opCtx,
+                                                                    bool dupsAllowed) {
+    return new UniqueV2BulkBuilder(this, opCtx, dupsAllowed, _prefix);
+}
+
 Status WiredTigerIndexUnique::_insert(WT_CURSOR* c,
                                       const BSONObj& key,
                                       const RecordId& id,
@@ -1185,6 +1359,73 @@ Status WiredTigerIndexUnique::_insert(WT_CURSOR* c,
     return wtRCToStatus(c->update(c));
 }
 
+/**
+ * This is duplicate of `WiredTigerIndexUnique::_insert` at the moment; might change later.
+ */
+Status WiredTigerIndexUniqueV2::_insert(WT_CURSOR* c,
+                                        const BSONObj& key,
+                                        const RecordId& id,
+                                        bool dupsAllowed) {
+    const KeyString data(keyStringVersion(), key, _ordering);
+    WiredTigerItem keyItem(data.getBuffer(), data.getSize());
+
+    KeyString value(keyStringVersion(), id);
+    if (!data.getTypeBits().isAllZeros())
+        value.appendTypeBits(data.getTypeBits());
+
+    WiredTigerItem valueItem(value.getBuffer(), value.getSize());
+    setKey(c, keyItem.Get());
+    c->set_value(c, valueItem.Get());
+    int ret = WT_OP_CHECK(c->insert(c));
+
+    if (ret != WT_DUPLICATE_KEY) {
+        return wtRCToStatus(ret);
+    }
+
+    // we might be in weird mode where there might be multiple values
+    // we put them all in the "list"
+    // Note that we can't omit AllZeros when there are multiple ids for a value. When we remove
+    // down to a single value, it will be cleaned up.
+    ret = WT_READ_CHECK(c->search(c));
+    invariantWTOK(ret);
+
+    WT_ITEM old;
+    invariantWTOK(c->get_value(c, &old));
+
+    bool insertedId = false;
+
+    value.resetToEmpty();
+    BufReader br(old.data, old.size);
+    while (br.remaining()) {
+        RecordId idInIndex = KeyString::decodeRecordId(&br);
+        if (id == idInIndex)
+            return Status::OK();  // already in index
+
+        if (!insertedId && id < idInIndex) {
+            value.appendRecordId(id);
+            value.appendTypeBits(data.getTypeBits());
+            insertedId = true;
+        }
+
+        // Copy from old to new value
+        value.appendRecordId(idInIndex);
+        value.appendTypeBits(KeyString::TypeBits::fromBuffer(keyStringVersion(), &br));
+    }
+
+    if (!dupsAllowed)
+        return dupKeyError(key);
+
+    if (!insertedId) {
+        // This id is higher than all currently in the index for this key
+        value.appendRecordId(id);
+        value.appendTypeBits(data.getTypeBits());
+    }
+
+    valueItem = WiredTigerItem(value.getBuffer(), value.getSize());
+    c->set_value(c, valueItem.Get());
+    return wtRCToStatus(c->update(c));
+}
+
 void WiredTigerIndexUnique::_unindex(WT_CURSOR* c,
                                      const BSONObj& key,
                                      const RecordId& id,
@@ -1290,6 +1531,113 @@ void WiredTigerIndexUnique::_unindex(WT_CURSOR* c,
     invariantWTOK(c->update(c));
 }
 
+/**
+ * This is duplicate of `WiredTigerIndexUnique::_unindex` at the moment; would change later.
+ */
+void WiredTigerIndexUniqueV2::_unindex(WT_CURSOR* c,
+                                       const BSONObj& key,
+                                       const RecordId& id,
+                                       bool dupsAllowed) {
+    KeyString data(keyStringVersion(), key, _ordering);
+    WiredTigerItem keyItem(data.getBuffer(), data.getSize());
+    setKey(c, keyItem.Get());
+
+    auto triggerWriteConflictAtPoint = [this, &keyItem](WT_CURSOR* point) {
+        // WT_NOTFOUND may occur during a background index build. Insert a dummy value and
+        // delete it again to trigger a write conflict in case this is being concurrently
+        // indexed by the background indexer.
+        setKey(point, keyItem.Get());
+        point->set_value(point, emptyItem.Get());
+        invariantWTOK(WT_OP_CHECK(point->insert(point)));
+        setKey(point, keyItem.Get());
+        invariantWTOK(WT_OP_CHECK(point->remove(point)));
+    };
+
+    if (!dupsAllowed) {
+        if (_partial) {
+            // Check that the record id matches.  We may be called to unindex records that are not
+            // present in the index due to the partial filter expression.
+            int ret = WT_READ_CHECK(c->search(c));
+            if (ret == WT_NOTFOUND) {
+                triggerWriteConflictAtPoint(c);
+                return;
+            }
+            invariantWTOK(ret);
+            WT_ITEM value;
+            invariantWTOK(c->get_value(c, &value));
+            BufReader br(value.data, value.size);
+            fassert(40691, br.remaining());
+            if (KeyString::decodeRecordId(&br) != id) {
+                return;
+            }
+            // Ensure there aren't any other values in here.
+            KeyString::TypeBits::fromBuffer(keyStringVersion(), &br);
+            fassert(40685, !br.remaining());
+        }
+        int ret = WT_OP_CHECK(c->remove(c));
+        if (ret == WT_NOTFOUND) {
+            return;
+        }
+        invariantWTOK(ret);
+        return;
+    }
+
+    // dups are allowed, so we have to deal with a vector of RecordIds.
+
+    int ret = WT_READ_CHECK(c->search(c));
+    if (ret == WT_NOTFOUND) {
+        triggerWriteConflictAtPoint(c);
+        return;
+    }
+    invariantWTOK(ret);
+
+    WT_ITEM old;
+    invariantWTOK(c->get_value(c, &old));
+
+    bool foundId = false;
+    std::vector<std::pair<RecordId, KeyString::TypeBits>> records;
+
+    BufReader br(old.data, old.size);
+    while (br.remaining()) {
+        RecordId idInIndex = KeyString::decodeRecordId(&br);
+        KeyString::TypeBits typeBits = KeyString::TypeBits::fromBuffer(keyStringVersion(), &br);
+
+        if (id == idInIndex) {
+            if (records.empty() && !br.remaining()) {
+                // This is the common case: we are removing the only id for this key.
+                // Remove the whole entry.
+                invariantWTOK(WT_OP_CHECK(c->remove(c)));
+                return;
+            }
+
+            foundId = true;
+            continue;
+        }
+
+        records.push_back(std::make_pair(idInIndex, typeBits));
+    }
+
+    if (!foundId) {
+        warning().stream() << id << " not found in the index for key " << redact(key);
+        return;  // nothing to do
+    }
+
+    // Put other ids for this key back in the index.
+    KeyString newValue(keyStringVersion());
+    invariant(!records.empty());
+    for (size_t i = 0; i < records.size(); i++) {
+        newValue.appendRecordId(records[i].first);
+        // When there is only one record, we can omit AllZeros TypeBits. Otherwise they need
+        // to be included.
+        if (!(records[i].second.isAllZeros() && records.size() == 1)) {
+            newValue.appendTypeBits(records[i].second);
+        }
+    }
+
+    WiredTigerItem valueItem = WiredTigerItem(newValue.getBuffer(), newValue.getSize());
+    c->set_value(c, valueItem.Get());
+    invariantWTOK(c->update(c));
+}
 // ------------------------------
 
 WiredTigerIndexStandard::WiredTigerIndexStandard(OperationContext* ctx,
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_index.h b/src/mongo/db/storage/wiredtiger/wiredtiger_index.h
index a90f770a529..8e37258aaf4 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_index.h
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_index.h
@@ -135,6 +135,11 @@ public:
 
     virtual bool unique() const = 0;
 
+    // Returns true if V2 unique index format is supported.
+    virtual bool isV2FormatUniqueIndex() const {
+        return false;
+    }
+
     Status dupKeyError(const BSONObj& key);
 
 protected:
@@ -153,6 +158,7 @@ protected:
     class BulkBuilder;
     class StandardBulkBuilder;
     class UniqueBulkBuilder;
+    class UniqueV2BulkBuilder;
 
     const Ordering _ordering;
     // The keystring version is effectively const after the WiredTigerIndex instance is constructed.
@@ -190,6 +196,35 @@ private:
     bool _partial;
 };
 
+class WiredTigerIndexUniqueV2 : public WiredTigerIndex {
+public:
+    WiredTigerIndexUniqueV2(OperationContext* ctx,
+                            const std::string& uri,
+                            const IndexDescriptor* desc,
+                            KVPrefix prefix,
+                            bool readOnly = false);
+
+    std::unique_ptr<SortedDataInterface::Cursor> newCursor(OperationContext* opCtx,
+                                                           bool forward) const override;
+
+    SortedDataBuilderInterface* getBulkBuilder(OperationContext* opCtx, bool dupsAllowed) override;
+
+    bool unique() const override {
+        return true;
+    }
+
+    bool isV2FormatUniqueIndex() const override {
+        return true;
+    }
+
+    Status _insert(WT_CURSOR* c, const BSONObj& key, const RecordId& id, bool dupsAllowed) override;
+
+    void _unindex(WT_CURSOR* c, const BSONObj& key, const RecordId& id, bool dupsAllowed) override;
+
+private:
+    bool _partial;
+};
+
 class WiredTigerIndexStandard : public WiredTigerIndex {
 public:
     WiredTigerIndexStandard(OperationContext* ctx,
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp
index aaa8472971e..d9adfe7c0f1 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp
@@ -781,8 +781,16 @@ SortedDataInterface* WiredTigerKVEngine::getGroupedSortedDataInterface(Operation
                                                                        StringData ident,
                                                                        const IndexDescriptor* desc,
                                                                        KVPrefix prefix) {
-    if (desc->unique())
-        return new WiredTigerIndexUnique(opCtx, _uri(ident), desc, prefix, _readOnly);
+    if (desc->unique()) {
+        // MongoDB 4.0 onwards new index version `kV2Unique` would be supported. By default unique
+        // index would be created with index version `kV2`. New format unique index would be created
+        // only if `IndexVersion` is `kV2Unique`.
+        if (desc->version() == IndexDescriptor::IndexVersion::kV2Unique)
+            return new WiredTigerIndexUniqueV2(opCtx, _uri(ident), desc, prefix, _readOnly);
+        else
+            return new WiredTigerIndexUnique(opCtx, _uri(ident), desc, prefix, _readOnly);
+    }
+
     return new WiredTigerIndexStandard(opCtx, _uri(ident), desc, prefix, _readOnly);
 }