SERVER-62793 Add index component to shard version

38 files changed, 533 insertions, 140 deletions

diff --git a/jstests/sharding/check_sharding_index_versioned.js b/jstests/sharding/check_sharding_index_versioned.js
index c47f2485dc9..2544e91d43f 100644
--- a/jstests/sharding/check_sharding_index_versioned.js
+++ b/jstests/sharding/check_sharding_index_versioned.js
@@ -21,7 +21,8 @@ assert.throwsWithCode(() => {
     st.rs0.getPrimary().getDB(dbName).runCommand({
         checkShardingIndex: ns,
         keyPattern: {x: 1},
-        shardVersion: {e: ObjectId(), t: Timestamp(1, 1), v: Timestamp(99, 10101)},
+        shardVersion:
+            {e: ObjectId(), t: Timestamp(1, 1), v: Timestamp(99, 10101), i: Timestamp(0, 0)},
     });
 }, ErrorCodes.StaleConfig);
 
diff --git a/jstests/sharding/libs/shard_versioning_util.js b/jstests/sharding/libs/shard_versioning_util.js
index 335c171a978..e918a6d172c 100644
--- a/jstests/sharding/libs/shard_versioning_util.js
+++ b/jstests/sharding/libs/shard_versioning_util.js
@@ -8,7 +8,8 @@ var ShardVersioningUtil = (function() {
     const kIgnoredShardVersion = {
         e: ObjectId("00000000ffffffffffffffff"),
         t: Timestamp(Math.pow(2, 32) - 1, Math.pow(2, 32) - 1),
-        v: Timestamp(0, 0)
+        v: Timestamp(0, 0),
+        i: Timestamp(0, 0)
     };
 
     /*
diff --git a/src/mongo/db/commands/run_aggregate.cpp b/src/mongo/db/commands/run_aggregate.cpp
index 01ed5a70a0a..981a692634e 100644
--- a/src/mongo/db/commands/run_aggregate.cpp
+++ b/src/mongo/db/commands/run_aggregate.cpp
@@ -885,7 +885,7 @@ Status runAggregate(OperationContext* opCtx,
             // This is prerequisite for future shard versioning checks.
             ScopedSetShardRole scopedSetShardRole(opCtx,
                                                   resolvedView.getNamespace(),
-                                                  ChunkVersion::UNSHARDED() /* shardVersion */,
+                                                  ShardVersion::UNSHARDED() /* shardVersion */,
                                                   boost::none /* databaseVersion */);
 
             uassert(std::move(resolvedView),
diff --git a/src/mongo/db/index_builds_coordinator_mongod.cpp b/src/mongo/db/index_builds_coordinator_mongod.cpp
index 12d4f6693c8..8f54e8fd9be 100644
--- a/src/mongo/db/index_builds_coordinator_mongod.cpp
+++ b/src/mongo/db/index_builds_coordinator_mongod.cpp
@@ -374,9 +374,7 @@ IndexBuildsCoordinatorMongod::_startIndexBuild(OperationContext* opCtx,
                                                  std::move(impersonatedClientAttrs.roleNames));
         }
 
-        boost::optional<ChunkVersion> chunkVersion =
-            shardVersion ? boost::make_optional((ChunkVersion)*shardVersion) : boost::none;
-        ScopedSetShardRole scopedSetShardRole(opCtx.get(), nss, chunkVersion, dbVersion);
+        ScopedSetShardRole scopedSetShardRole(opCtx.get(), nss, shardVersion, dbVersion);
 
         {
             stdx::unique_lock<Client> lk(*opCtx->getClient());
diff --git a/src/mongo/db/pipeline/process_interface/shardsvr_process_interface.cpp b/src/mongo/db/pipeline/process_interface/shardsvr_process_interface.cpp
index 2acfd4c2012..837248fe720 100644
--- a/src/mongo/db/pipeline/process_interface/shardsvr_process_interface.cpp
+++ b/src/mongo/db/pipeline/process_interface/shardsvr_process_interface.cpp
@@ -400,7 +400,7 @@ ShardServerProcessInterface::expectUnshardedCollectionInScope(
         ScopedExpectUnshardedCollectionImpl(OperationContext* opCtx,
                                             const NamespaceString& nss,
                                             const boost::optional<DatabaseVersion>& dbVersion)
-            : _expectUnsharded(opCtx, nss, ChunkVersion::UNSHARDED(), dbVersion) {}
+            : _expectUnsharded(opCtx, nss, ShardVersion::UNSHARDED(), dbVersion) {}
 
     private:
         ScopedSetShardRole _expectUnsharded;
diff --git a/src/mongo/db/s/balancer/balancer_commands_scheduler.h b/src/mongo/db/s/balancer/balancer_commands_scheduler.h
index e63d40bf483..bf2ad14bb7c 100644
--- a/src/mongo/db/s/balancer/balancer_commands_scheduler.h
+++ b/src/mongo/db/s/balancer/balancer_commands_scheduler.h
@@ -145,7 +145,7 @@ public:
                                                          const NamespaceString& nss,
                                                          const ShardId& shardId,
                                                          const ChunkRange& chunkRange,
-                                                         const ChunkVersion& version,
+                                                         const ShardVersion& version,
                                                          const KeyPattern& keyPattern,
                                                          bool estimatedValue) = 0;
 
diff --git a/src/mongo/db/s/balancer/balancer_commands_scheduler_impl.cpp b/src/mongo/db/s/balancer/balancer_commands_scheduler_impl.cpp
index 88300a97779..f2a2ca609dd 100644
--- a/src/mongo/db/s/balancer/balancer_commands_scheduler_impl.cpp
+++ b/src/mongo/db/s/balancer/balancer_commands_scheduler_impl.cpp
@@ -357,7 +357,7 @@ SemiFuture<DataSizeResponse> BalancerCommandsSchedulerImpl::requestDataSize(
     const NamespaceString& nss,
     const ShardId& shardId,
     const ChunkRange& chunkRange,
-    const ChunkVersion& version,
+    const ShardVersion& version,
     const KeyPattern& keyPattern,
     bool estimatedValue) {
     auto commandInfo = std::make_shared<DataSizeCommandInfo>(nss,
diff --git a/src/mongo/db/s/balancer/balancer_commands_scheduler_impl.h b/src/mongo/db/s/balancer/balancer_commands_scheduler_impl.h
index 04977417db0..0d36e2d5959 100644
--- a/src/mongo/db/s/balancer/balancer_commands_scheduler_impl.h
+++ b/src/mongo/db/s/balancer/balancer_commands_scheduler_impl.h
@@ -337,7 +337,7 @@ public:
                         const BSONObj& lowerBoundKey,
                         const BSONObj& upperBoundKey,
                         bool estimatedValue,
-                        const ChunkVersion& version)
+                        const ShardVersion& version)
         : CommandInfo(shardId, nss, boost::none),
           _shardKeyPattern(shardKeyPattern),
           _lowerBoundKey(lowerBoundKey),
@@ -353,7 +353,7 @@ public:
             .append(kMaxValue, _upperBoundKey)
             .append(kEstimatedValue, _estimatedValue);
 
-        _version.serialize(ChunkVersion::kChunkVersionField, &commandBuilder);
+        _version.serialize(ShardVersion::kShardVersionField, &commandBuilder);
 
         return commandBuilder.obj();
     }
@@ -363,7 +363,7 @@ private:
     BSONObj _lowerBoundKey;
     BSONObj _upperBoundKey;
     bool _estimatedValue;
-    ChunkVersion _version;
+    ShardVersion _version;
 
     static const std::string kCommandName;
     static const std::string kKeyPattern;
@@ -577,7 +577,7 @@ public:
                                                  const NamespaceString& nss,
                                                  const ShardId& shardId,
                                                  const ChunkRange& chunkRange,
-                                                 const ChunkVersion& version,
+                                                 const ShardVersion& version,
                                                  const KeyPattern& keyPattern,
                                                  bool estimatedValue) override;
 
diff --git a/src/mongo/db/s/balancer/balancer_policy.cpp b/src/mongo/db/s/balancer/balancer_policy.cpp
index 8c2167e6b2b..07ed023e774 100644
--- a/src/mongo/db/s/balancer/balancer_policy.cpp
+++ b/src/mongo/db/s/balancer/balancer_policy.cpp
@@ -944,7 +944,7 @@ DataSizeInfo::DataSizeInfo(const ShardId& shardId,
                            const NamespaceString& nss,
                            const UUID& uuid,
                            const ChunkRange& chunkRange,
-                           const ChunkVersion& version,
+                           const ShardVersion& version,
                            const KeyPattern& keyPattern,
                            bool estimatedValue)
     : shardId(shardId),
diff --git a/src/mongo/db/s/balancer/balancer_policy.h b/src/mongo/db/s/balancer/balancer_policy.h
index 5759a319ac9..e88bc0a67fc 100644
--- a/src/mongo/db/s/balancer/balancer_policy.h
+++ b/src/mongo/db/s/balancer/balancer_policy.h
@@ -41,6 +41,7 @@
 #include "mongo/s/request_types/auto_split_vector_gen.h"
 #include "mongo/s/request_types/move_range_request_gen.h"
 #include "mongo/s/shard_id.h"
+#include "mongo/s/shard_version.h"
 
 namespace mongo {
 
@@ -181,7 +182,7 @@ struct DataSizeInfo {
                  const NamespaceString& nss,
                  const UUID& uuid,
                  const ChunkRange& chunkRange,
-                 const ChunkVersion& version,
+                 const ShardVersion& version,
                  const KeyPattern& keyPattern,
                  bool estimatedValue);
 
@@ -189,7 +190,9 @@ struct DataSizeInfo {
     NamespaceString nss;
     UUID uuid;
     ChunkRange chunkRange;
-    ChunkVersion version;
+    // Use ShardVersion for CRUD targeting since datasize is considered a CRUD operation, not a DDL
+    // operation.
+    ShardVersion version;
     KeyPattern keyPattern;
     bool estimatedValue;
 };
diff --git a/src/mongo/db/s/collection_metadata_filtering_test.cpp b/src/mongo/db/s/collection_metadata_filtering_test.cpp
index 74dc6a9e655..497bcba54cf 100644
--- a/src/mongo/db/s/collection_metadata_filtering_test.cpp
+++ b/src/mongo/db/s/collection_metadata_filtering_test.cpp
@@ -148,10 +148,11 @@ TEST_F(CollectionMetadataFilteringTest, FilterDocumentsInTheFuture) {
     ASSERT_OK(readConcernArgs.initialize(readConcern["readConcern"]));
 
     AutoGetCollection autoColl(operationContext(), kNss, MODE_IS);
-    ScopedSetShardRole scopedSetShardRole{operationContext(),
-                                          kNss,
-                                          metadata.getShardVersion() /* shardVersion */,
-                                          boost::none /* databaseVersion */};
+    ScopedSetShardRole scopedSetShardRole{
+        operationContext(),
+        kNss,
+        ShardVersion(metadata.getShardVersion()) /* shardVersion */,
+        boost::none /* databaseVersion */};
     auto* const css = CollectionShardingState::get(operationContext(), kNss);
     testFilterFn(css->getOwnershipFilter(
         operationContext(), CollectionShardingState::OrphanCleanupPolicy::kAllowOrphanCleanup));
@@ -176,10 +177,11 @@ TEST_F(CollectionMetadataFilteringTest, FilterDocumentsInThePast) {
     ASSERT_OK(readConcernArgs.initialize(readConcern["readConcern"]));
 
     AutoGetCollection autoColl(operationContext(), kNss, MODE_IS);
-    ScopedSetShardRole scopedSetShardRole{operationContext(),
-                                          kNss,
-                                          metadata.getShardVersion() /* shardVersion */,
-                                          boost::none /* databaseVersion */};
+    ScopedSetShardRole scopedSetShardRole{
+        operationContext(),
+        kNss,
+        ShardVersion(metadata.getShardVersion()) /* shardVersion */,
+        boost::none /* databaseVersion */};
     auto* const css = CollectionShardingState::get(operationContext(), kNss);
     testFilterFn(css->getOwnershipFilter(
         operationContext(), CollectionShardingState::OrphanCleanupPolicy::kAllowOrphanCleanup));
@@ -212,10 +214,11 @@ TEST_F(CollectionMetadataFilteringTest, FilterDocumentsTooFarInThePastThrowsStal
     ASSERT_OK(readConcernArgs.initialize(readConcern["readConcern"]));
 
     AutoGetCollection autoColl(operationContext(), kNss, MODE_IS);
-    ScopedSetShardRole scopedSetShardRole{operationContext(),
-                                          kNss,
-                                          metadata.getShardVersion() /* shardVersion */,
-                                          boost::none /* databaseVersion */};
+    ScopedSetShardRole scopedSetShardRole{
+        operationContext(),
+        kNss,
+        ShardVersion(metadata.getShardVersion()) /* shardVersion */,
+        boost::none /* databaseVersion */};
     auto* const css = CollectionShardingState::get(operationContext(), kNss);
     testFilterFn(css->getOwnershipFilter(
         operationContext(), CollectionShardingState::OrphanCleanupPolicy::kAllowOrphanCleanup));
diff --git a/src/mongo/db/s/collection_sharding_runtime.cpp b/src/mongo/db/s/collection_sharding_runtime.cpp
index 74ec459230d..bf444811915 100644
--- a/src/mongo/db/s/collection_sharding_runtime.cpp
+++ b/src/mongo/db/s/collection_sharding_runtime.cpp
@@ -40,6 +40,7 @@
 #include "mongo/db/s/sharding_runtime_d_params_gen.h"
 #include "mongo/db/s/sharding_state.h"
 #include "mongo/logv2/log.h"
+#include "mongo/s/sharding_feature_flags_gen.h"
 #include "mongo/s/type_collection_common_types_gen.h"
 #include "mongo/util/duration.h"
 
@@ -109,6 +110,12 @@ ScopedCollectionFilter CollectionShardingRuntime::getOwnershipFilter(
         // No operations should be calling getOwnershipFilter without a shard version
         invariant(optReceivedShardVersion,
                   "getOwnershipFilter called by operation that doesn't specify shard version");
+        uassert(6279300,
+                "Request was received without an attached index version. This could indicate that "
+                "this request was sent by a router of an older version",
+                !feature_flags::gGlobalIndexesShardingCatalog.isEnabled(
+                    serverGlobalParams.featureCompatibility) ||
+                    optReceivedShardVersion->indexVersion());
     }
 
     auto metadata =
@@ -141,6 +148,12 @@ ScopedCollectionDescription CollectionShardingRuntime::getCollectionDescription(
 
     auto optMetadata = _getCurrentMetadataIfKnown(boost::none);
     const auto receivedShardVersion{oss.getShardVersion(_nss)};
+    uassert(6279301,
+            "Request was received without an attached index version. This could indicate that this "
+            "request was sent by a router of an older version",
+            !feature_flags::gGlobalIndexesShardingCatalog.isEnabled(
+                serverGlobalParams.featureCompatibility) ||
+                !receivedShardVersion || receivedShardVersion->indexVersion());
     uassert(
         StaleConfigInfo(_nss,
                         receivedShardVersion ? (ChunkVersion)*receivedShardVersion
@@ -355,6 +368,13 @@ CollectionShardingRuntime::_getMetadataWithVersionCheckAt(
     // Assume that the received shard version was IGNORED if the current operation wasn't versioned
     const auto& receivedShardVersion =
         optReceivedShardVersion ? (ChunkVersion)*optReceivedShardVersion : ChunkVersion::IGNORED();
+    uassert(6279302,
+            "Request was received without an attached index version. This could indicate that this "
+            "request was sent by a router of an older version",
+            !feature_flags::gGlobalIndexesShardingCatalog.isEnabled(
+                serverGlobalParams.featureCompatibility) ||
+                receivedShardVersion == ChunkVersion::IGNORED() ||
+                optReceivedShardVersion->indexVersion());
 
     auto csrLock = CSRLock::lockShared(opCtx, this);
 
diff --git a/src/mongo/db/s/collection_sharding_runtime_test.cpp b/src/mongo/db/s/collection_sharding_runtime_test.cpp
index c4884089054..4f7c9560ab1 100644
--- a/src/mongo/db/s/collection_sharding_runtime_test.cpp
+++ b/src/mongo/db/s/collection_sharding_runtime_test.cpp
@@ -88,8 +88,10 @@ TEST_F(CollectionShardingRuntimeTest,
     CollectionShardingRuntime csr(getServiceContext(), kTestNss, executor());
     ASSERT_FALSE(csr.getCollectionDescription(opCtx).isSharded());
     auto metadata = makeShardedMetadata(opCtx);
-    ScopedSetShardRole scopedSetShardRole{
-        opCtx, kTestNss, metadata.getShardVersion(), boost::none /* databaseVersion */};
+    ScopedSetShardRole scopedSetShardRole{opCtx,
+                                          kTestNss,
+                                          ShardVersion(metadata.getShardVersion()),
+                                          boost::none /* databaseVersion */};
     ASSERT_THROWS_CODE(csr.getCollectionDescription(opCtx), DBException, ErrorCodes::StaleConfig);
 }
 
@@ -107,8 +109,10 @@ TEST_F(CollectionShardingRuntimeTest,
     OperationContext* opCtx = operationContext();
     auto metadata = makeShardedMetadata(opCtx);
     csr.setFilteringMetadata(opCtx, metadata);
-    ScopedSetShardRole scopedSetShardRole{
-        opCtx, kTestNss, metadata.getShardVersion(), boost::none /* databaseVersion */};
+    ScopedSetShardRole scopedSetShardRole{opCtx,
+                                          kTestNss,
+                                          ShardVersion(metadata.getShardVersion()),
+                                          boost::none /* databaseVersion */};
     ASSERT_TRUE(csr.getCollectionDescription(opCtx).isSharded());
 }
 
@@ -171,8 +175,10 @@ TEST_F(CollectionShardingRuntimeTest,
     OperationContext* opCtx = operationContext();
     auto metadata = makeShardedMetadata(opCtx);
     csr.setFilteringMetadata(opCtx, metadata);
-    ScopedSetShardRole scopedSetShardRole{
-        opCtx, kTestNss, metadata.getShardVersion(), boost::none /* databaseVersion */};
+    ScopedSetShardRole scopedSetShardRole{opCtx,
+                                          kTestNss,
+                                          ShardVersion(metadata.getShardVersion()),
+                                          boost::none /* databaseVersion */};
     ASSERT_EQ(csr.getNumMetadataManagerChanges_forTest(), 1);
 
     // Set it again with a different metadata object (UUID is generated randomly in
@@ -200,7 +206,7 @@ TEST_F(CollectionShardingRuntimeTest, ReturnUnshardedMetadataInServerlessMode) {
     ScopedSetShardRole scopedSetShardRole1{
         opCtx,
         testNss,
-        ChunkVersion::UNSHARDED(), /* shardVersion */
+        ShardVersion::UNSHARDED(), /* shardVersion */
         boost::none                /* databaseVersion */
     };
 
@@ -215,8 +221,8 @@ TEST_F(CollectionShardingRuntimeTest, ReturnUnshardedMetadataInServerlessMode) {
     ScopedSetShardRole scopedSetShardRole2{
         opCtx,
         NamespaceString::kLogicalSessionsNamespace,
-        ChunkVersion({OID::gen(), Timestamp(1, 1)}, {1, 0}), /* shardVersion */
-        boost::none                                          /* databaseVersion */
+        ShardVersion(ChunkVersion({OID::gen(), Timestamp(1, 1)}, {1, 0})), /* shardVersion */
+        boost::none                                                        /* databaseVersion */
     };
 
     CollectionShardingRuntime csrLogicalSession(
diff --git a/src/mongo/db/s/op_observer_sharding_test.cpp b/src/mongo/db/s/op_observer_sharding_test.cpp
index 19efe36166a..628a62ae180 100644
--- a/src/mongo/db/s/op_observer_sharding_test.cpp
+++ b/src/mongo/db/s/op_observer_sharding_test.cpp
@@ -106,10 +106,11 @@ TEST_F(DocumentKeyStateTest, MakeDocumentKeyStateShardedWithoutIdInShardKey) {
     setCollectionFilteringMetadata(operationContext(), metadata);
 
     AutoGetCollection autoColl(operationContext(), kTestNss, MODE_IX);
-    ScopedSetShardRole scopedSetShardRole{operationContext(),
-                                          kTestNss,
-                                          metadata.getShardVersion() /* shardVersion */,
-                                          boost::none /* databaseVersion */};
+    ScopedSetShardRole scopedSetShardRole{
+        operationContext(),
+        kTestNss,
+        ShardVersion(metadata.getShardVersion()) /* shardVersion */,
+        boost::none /* databaseVersion */};
 
     // The order of fields in `doc` deliberately does not match the shard key
     auto doc = BSON("key3"
@@ -133,10 +134,11 @@ TEST_F(DocumentKeyStateTest, MakeDocumentKeyStateShardedWithIdInShardKey) {
     setCollectionFilteringMetadata(operationContext(), metadata);
 
     AutoGetCollection autoColl(operationContext(), kTestNss, MODE_IX);
-    ScopedSetShardRole scopedSetShardRole{operationContext(),
-                                          kTestNss,
-                                          metadata.getShardVersion() /* shardVersion */,
-                                          boost::none /* databaseVersion */};
+    ScopedSetShardRole scopedSetShardRole{
+        operationContext(),
+        kTestNss,
+        ShardVersion(metadata.getShardVersion()) /* shardVersion */,
+        boost::none /* databaseVersion */};
 
     // The order of fields in `doc` deliberately does not match the shard key
     auto doc = BSON("key2" << true << "key3"
@@ -160,10 +162,11 @@ TEST_F(DocumentKeyStateTest, MakeDocumentKeyStateShardedWithIdHashInShardKey) {
     setCollectionFilteringMetadata(operationContext(), metadata);
 
     AutoGetCollection autoColl(operationContext(), kTestNss, MODE_IX);
-    ScopedSetShardRole scopedSetShardRole{operationContext(),
-                                          kTestNss,
-                                          metadata.getShardVersion() /* shardVersion */,
-                                          boost::none /* databaseVersion */};
+    ScopedSetShardRole scopedSetShardRole{
+        operationContext(),
+        kTestNss,
+        ShardVersion(metadata.getShardVersion()) /* shardVersion */,
+        boost::none /* databaseVersion */};
 
     auto doc = BSON("key2" << true << "_id"
                            << "hello"
diff --git a/src/mongo/db/s/operation_sharding_state.cpp b/src/mongo/db/s/operation_sharding_state.cpp
index 0621a5bd87a..f35268f8703 100644
--- a/src/mongo/db/s/operation_sharding_state.cpp
+++ b/src/mongo/db/s/operation_sharding_state.cpp
@@ -56,19 +56,18 @@ bool OperationShardingState::isComingFromRouter(OperationContext* opCtx) {
 
 void OperationShardingState::setShardRole(OperationContext* opCtx,
                                           const NamespaceString& nss,
-                                          const boost::optional<ChunkVersion>& shardVersion,
+                                          const boost::optional<ShardVersion>& shardVersion,
                                           const boost::optional<DatabaseVersion>& databaseVersion) {
     auto& oss = OperationShardingState::get(opCtx);
 
     if (shardVersion) {
-        ShardVersion fullShardVersion(*shardVersion);
-        auto emplaceResult = oss._shardVersions.try_emplace(nss.ns(), fullShardVersion);
+        auto emplaceResult = oss._shardVersions.try_emplace(nss.ns(), *shardVersion);
         auto& tracker = emplaceResult.first->second;
         if (!emplaceResult.second) {
             uassert(640570,
                     str::stream() << "Illegal attempt to change the expected shard version for "
-                                  << nss << " from " << tracker.v << " to " << fullShardVersion,
-                    tracker.v == fullShardVersion);
+                                  << nss << " from " << tracker.v << " to " << *shardVersion,
+                    tracker.v == *shardVersion);
         }
         invariant(++tracker.recursion > 0);
     }
@@ -188,7 +187,7 @@ ScopedAllowImplicitCollectionCreate_UNSAFE::~ScopedAllowImplicitCollectionCreate
 
 ScopedSetShardRole::ScopedSetShardRole(OperationContext* opCtx,
                                        NamespaceString nss,
-                                       boost::optional<ChunkVersion> shardVersion,
+                                       boost::optional<ShardVersion> shardVersion,
                                        boost::optional<DatabaseVersion> databaseVersion)
     : _opCtx(opCtx),
       _nss(std::move(nss)),
diff --git a/src/mongo/db/s/operation_sharding_state.h b/src/mongo/db/s/operation_sharding_state.h
index 695b33f4927..22f5993adaf 100644
--- a/src/mongo/db/s/operation_sharding_state.h
+++ b/src/mongo/db/s/operation_sharding_state.h
@@ -49,7 +49,7 @@ class ScopedSetShardRole {
 public:
     ScopedSetShardRole(OperationContext* opCtx,
                        NamespaceString nss,
-                       boost::optional<ChunkVersion> shardVersion,
+                       boost::optional<ShardVersion> shardVersion,
                        boost::optional<DatabaseVersion> databaseVersion);
     ~ScopedSetShardRole();
 
@@ -58,7 +58,7 @@ private:
 
     NamespaceString _nss;
 
-    boost::optional<ChunkVersion> _shardVersion;
+    boost::optional<ShardVersion> _shardVersion;
     boost::optional<DatabaseVersion> _databaseVersion;
 };
 
@@ -113,7 +113,7 @@ public:
      */
     static void setShardRole(OperationContext* opCtx,
                              const NamespaceString& nss,
-                             const boost::optional<ChunkVersion>& shardVersion,
+                             const boost::optional<ShardVersion>& shardVersion,
                              const boost::optional<DatabaseVersion>& dbVersion);
 
     /**
diff --git a/src/mongo/db/s/operation_sharding_state_test.cpp b/src/mongo/db/s/operation_sharding_state_test.cpp
index 9c275398f85..a85ebdde57a 100644
--- a/src/mongo/db/s/operation_sharding_state_test.cpp
+++ b/src/mongo/db/s/operation_sharding_state_test.cpp
@@ -48,7 +48,8 @@ TEST_F(OperationShardingStateTest, ScopedSetShardRoleDbVersion) {
 
 TEST_F(OperationShardingStateTest, ScopedSetShardRoleShardVersion) {
     ChunkVersion shardVersion({OID::gen(), Timestamp(1, 0)}, {1, 0});
-    ScopedSetShardRole scopedSetShardRole(operationContext(), kNss, shardVersion, boost::none);
+    ScopedSetShardRole scopedSetShardRole(
+        operationContext(), kNss, ShardVersion(shardVersion), boost::none);
 
     auto& oss = OperationShardingState::get(operationContext());
     ASSERT_EQ(shardVersion, *oss.getShardVersion(kNss));
@@ -60,13 +61,13 @@ TEST_F(OperationShardingStateTest, ScopedSetShardRoleChangeShardVersionSameNames
     {
         ChunkVersion shardVersion1({OID::gen(), Timestamp(10, 0)}, {1, 0});
         ScopedSetShardRole scopedSetShardRole1(
-            operationContext(), kNss, shardVersion1, boost::none);
+            operationContext(), kNss, ShardVersion(shardVersion1), boost::none);
         ASSERT_EQ(shardVersion1, *oss.getShardVersion(kNss));
     }
     {
         ChunkVersion shardVersion2({OID::gen(), Timestamp(20, 0)}, {1, 0});
         ScopedSetShardRole scopedSetShardRole2(
-            operationContext(), kNss, shardVersion2, boost::none);
+            operationContext(), kNss, ShardVersion(shardVersion2), boost::none);
         ASSERT_EQ(shardVersion2, *oss.getShardVersion(kNss));
     }
 }
@@ -75,9 +76,10 @@ TEST_F(OperationShardingStateTest, ScopedSetShardRoleRecursiveShardVersionDiffer
     ChunkVersion shardVersion1({OID::gen(), Timestamp(10, 0)}, {1, 0});
     ChunkVersion shardVersion2({OID::gen(), Timestamp(20, 0)}, {1, 0});
 
-    ScopedSetShardRole scopedSetShardRole1(operationContext(), kNss, shardVersion1, boost::none);
+    ScopedSetShardRole scopedSetShardRole1(
+        operationContext(), kNss, ShardVersion(shardVersion1), boost::none);
     ScopedSetShardRole scopedSetShardRole2(
-        operationContext(), kAnotherNss, shardVersion2, boost::none);
+        operationContext(), kAnotherNss, ShardVersion(shardVersion2), boost::none);
 
     auto& oss = OperationShardingState::get(operationContext());
     ASSERT_EQ(shardVersion1, *oss.getShardVersion(kNss));
diff --git a/src/mongo/db/s/resharding/resharding_collection_cloner.cpp b/src/mongo/db/s/resharding/resharding_collection_cloner.cpp
index fffac5d1736..d72826aabb4 100644
--- a/src/mongo/db/s/resharding/resharding_collection_cloner.cpp
+++ b/src/mongo/db/s/resharding/resharding_collection_cloner.cpp
@@ -283,7 +283,7 @@ bool ReshardingCollectionCloner::doOneBatch(OperationContext* opCtx, Pipeline& p
     // recovered.
     ScopedSetShardRole scopedSetShardRole(opCtx,
                                           _outputNss,
-                                          ChunkVersion::IGNORED() /* shardVersion */,
+                                          ShardVersion::IGNORED() /* shardVersion */,
                                           boost::none /* databaseVersion */);
 
     Timer batchInsertTimer;
diff --git a/src/mongo/db/s/resharding/resharding_destined_recipient_test.cpp b/src/mongo/db/s/resharding/resharding_destined_recipient_test.cpp
index 8d29958aa9d..fbaa3ddedc8 100644
--- a/src/mongo/db/s/resharding/resharding_destined_recipient_test.cpp
+++ b/src/mongo/db/s/resharding/resharding_destined_recipient_test.cpp
@@ -176,7 +176,7 @@ protected:
         NamespaceString tempNss;
         UUID sourceUuid;
         ShardId destShard;
-        ChunkVersion version;
+        ShardVersion version;
         DatabaseVersion dbVersion{UUID::gen(), Timestamp(1, 1)};
     };
 
@@ -196,7 +196,7 @@ protected:
 
         ReshardingEnv env(CollectionCatalog::get(opCtx)->lookupUUIDByNSS(opCtx, kNss).value());
         env.destShard = kShardList[1].getName();
-        env.version = ChunkVersion({OID::gen(), Timestamp(1, 1)}, {1, 0});
+        env.version = ShardVersion(ChunkVersion({OID::gen(), Timestamp(1, 1)}, {1, 0}));
         env.tempNss =
             NamespaceString(kNss.db(),
                             fmt::format("{}{}",
@@ -391,7 +391,7 @@ TEST_F(DestinedRecipientTest, TestOpObserverSetsDestinedRecipientOnMultiUpdates)
 
     auto env = setupReshardingEnv(opCtx, true);
 
-    OperationShardingState::setShardRole(opCtx, kNss, ChunkVersion::IGNORED(), env.dbVersion);
+    OperationShardingState::setShardRole(opCtx, kNss, ShardVersion::IGNORED(), env.dbVersion);
     client.update(kNss.ns(),
                   BSON("x" << 0),
                   BSON("$set" << BSON("z" << 5)),
diff --git a/src/mongo/db/s/resharding/resharding_donor_recipient_common_test.cpp b/src/mongo/db/s/resharding/resharding_donor_recipient_common_test.cpp
index 7068918b875..c98f6352322 100644
--- a/src/mongo/db/s/resharding/resharding_donor_recipient_common_test.cpp
+++ b/src/mongo/db/s/resharding/resharding_donor_recipient_common_test.cpp
@@ -268,10 +268,11 @@ protected:
     void addFilteringMetadata(OperationContext* opCtx, NamespaceString sourceNss, ShardId shardId) {
         AutoGetCollection autoColl(opCtx, sourceNss, LockMode::MODE_IS);
         const auto metadata{makeShardedMetadataForOriginalCollection(opCtx, shardId)};
-        ScopedSetShardRole scopedSetShardRole{opCtx,
-                                              sourceNss,
-                                              metadata.getShardVersion() /* shardVersion */,
-                                              boost::none /* databaseVersion */};
+        ScopedSetShardRole scopedSetShardRole{
+            opCtx,
+            sourceNss,
+            ShardVersion(metadata.getShardVersion()) /* shardVersion */,
+            boost::none /* databaseVersion */};
 
         auto csr = CollectionShardingRuntime::get(opCtx, sourceNss);
         csr->setFilteringMetadata(opCtx, metadata);
@@ -349,10 +350,11 @@ protected:
 TEST_F(ReshardingDonorRecipientCommonInternalsTest, ConstructDonorDocumentFromReshardingFields) {
     OperationContext* opCtx = operationContext();
     auto metadata = makeShardedMetadataForOriginalCollection(opCtx, kThisShard.getShardId());
-    ScopedSetShardRole scopedSetShardRole{opCtx,
-                                          kOriginalNss,
-                                          metadata.getShardVersion() /* shardVersion */,
-                                          boost::none /* databaseVersion */};
+    ScopedSetShardRole scopedSetShardRole{
+        opCtx,
+        kOriginalNss,
+        ShardVersion(metadata.getShardVersion()) /* shardVersion */,
+        boost::none /* databaseVersion */};
 
     auto reshardingFields =
         createCommonReshardingFields(kReshardingUUID, CoordinatorStateEnum::kPreparingToDonate);
@@ -368,10 +370,11 @@ TEST_F(ReshardingDonorRecipientCommonInternalsTest,
     OperationContext* opCtx = operationContext();
     auto metadata =
         makeShardedMetadataForTemporaryReshardingCollection(opCtx, kThisShard.getShardId());
-    ScopedSetShardRole scopedSetShardRole{opCtx,
-                                          kTemporaryReshardingNss,
-                                          metadata.getShardVersion() /* shardVersion */,
-                                          boost::none /* databaseVersion */};
+    ScopedSetShardRole scopedSetShardRole{
+        opCtx,
+        kTemporaryReshardingNss,
+        ShardVersion(metadata.getShardVersion()) /* shardVersion */,
+        boost::none /* databaseVersion */};
 
     auto reshardingFields =
         createCommonReshardingFields(kReshardingUUID, CoordinatorStateEnum::kPreparingToDonate);
@@ -386,10 +389,11 @@ TEST_F(ReshardingDonorRecipientCommonInternalsTest,
 TEST_F(ReshardingDonorRecipientCommonTest, CreateDonorServiceInstance) {
     OperationContext* opCtx = operationContext();
     auto metadata = makeShardedMetadataForOriginalCollection(opCtx, kThisShard.getShardId());
-    ScopedSetShardRole scopedSetShardRole{opCtx,
-                                          kOriginalNss,
-                                          metadata.getShardVersion() /* shardVersion */,
-                                          boost::none /* databaseVersion */};
+    ScopedSetShardRole scopedSetShardRole{
+        opCtx,
+        kOriginalNss,
+        ShardVersion(metadata.getShardVersion()) /* shardVersion */,
+        boost::none /* databaseVersion */};
 
     auto reshardingFields =
         createCommonReshardingFields(kReshardingUUID, CoordinatorStateEnum::kPreparingToDonate);
@@ -412,10 +416,11 @@ TEST_F(ReshardingDonorRecipientCommonTest, CreateRecipientServiceInstance) {
     OperationContext* opCtx = operationContext();
     auto metadata =
         makeShardedMetadataForTemporaryReshardingCollection(opCtx, kThisShard.getShardId());
-    ScopedSetShardRole scopedSetShardRole{opCtx,
-                                          kTemporaryReshardingNss,
-                                          metadata.getShardVersion() /* shardVersion */,
-                                          boost::none /* databaseVersion */};
+    ScopedSetShardRole scopedSetShardRole{
+        opCtx,
+        kTemporaryReshardingNss,
+        ShardVersion(metadata.getShardVersion()) /* shardVersion */,
+        boost::none /* databaseVersion */};
 
     auto reshardingFields =
         createCommonReshardingFields(kReshardingUUID, CoordinatorStateEnum::kPreparingToDonate);
@@ -439,10 +444,11 @@ TEST_F(ReshardingDonorRecipientCommonTest,
        CreateDonorServiceInstanceWithIncorrectCoordinatorState) {
     OperationContext* opCtx = operationContext();
     auto metadata = makeShardedMetadataForOriginalCollection(opCtx, kThisShard.getShardId());
-    ScopedSetShardRole scopedSetShardRole{opCtx,
-                                          kOriginalNss,
-                                          metadata.getShardVersion() /* shardVersion */,
-                                          boost::none /* databaseVersion */};
+    ScopedSetShardRole scopedSetShardRole{
+        opCtx,
+        kOriginalNss,
+        ShardVersion(metadata.getShardVersion()) /* shardVersion */,
+        boost::none /* databaseVersion */};
 
     auto reshardingFields =
         createCommonReshardingFields(kReshardingUUID, CoordinatorStateEnum::kCommitting);
@@ -460,10 +466,11 @@ TEST_F(ReshardingDonorRecipientCommonTest,
     OperationContext* opCtx = operationContext();
     auto metadata =
         makeShardedMetadataForTemporaryReshardingCollection(opCtx, kThisShard.getShardId());
-    ScopedSetShardRole scopedSetShardRole{opCtx,
-                                          kTemporaryReshardingNss,
-                                          metadata.getShardVersion() /* shardVersion */,
-                                          boost::none /* databaseVersion */};
+    ScopedSetShardRole scopedSetShardRole{
+        opCtx,
+        kTemporaryReshardingNss,
+        ShardVersion(metadata.getShardVersion()) /* shardVersion */,
+        boost::none /* databaseVersion */};
 
     auto reshardingFields =
         createCommonReshardingFields(kReshardingUUID, CoordinatorStateEnum::kCommitting);
@@ -487,10 +494,11 @@ TEST_F(ReshardingDonorRecipientCommonTest,
 TEST_F(ReshardingDonorRecipientCommonTest, ProcessDonorFieldsWhenShardDoesntOwnAnyChunks) {
     OperationContext* opCtx = operationContext();
     auto metadata = makeShardedMetadataForOriginalCollection(opCtx, kOtherShard.getShardId());
-    ScopedSetShardRole scopedSetShardRole{opCtx,
-                                          kOriginalNss,
-                                          metadata.getShardVersion() /* shardVersion */,
-                                          boost::none /* databaseVersion */};
+    ScopedSetShardRole scopedSetShardRole{
+        opCtx,
+        kOriginalNss,
+        ShardVersion(metadata.getShardVersion()) /* shardVersion */,
+        boost::none /* databaseVersion */};
 
     auto reshardingFields =
         createCommonReshardingFields(kReshardingUUID, CoordinatorStateEnum::kPreparingToDonate);
@@ -511,10 +519,11 @@ TEST_F(ReshardingDonorRecipientCommonTest, ProcessRecipientFieldsWhenShardDoesnt
     OperationContext* opCtx = operationContext();
     auto metadata =
         makeShardedMetadataForTemporaryReshardingCollection(opCtx, kOtherShard.getShardId());
-    ScopedSetShardRole scopedSetShardRole{opCtx,
-                                          kTemporaryReshardingNss,
-                                          metadata.getShardVersion() /* shardVersion */,
-                                          boost::none /* databaseVersion */};
+    ScopedSetShardRole scopedSetShardRole{
+        opCtx,
+        kTemporaryReshardingNss,
+        ShardVersion(metadata.getShardVersion()) /* shardVersion */,
+        boost::none /* databaseVersion */};
 
     auto reshardingFields =
         createCommonReshardingFields(kReshardingUUID, CoordinatorStateEnum::kPreparingToDonate);
@@ -536,10 +545,11 @@ TEST_F(ReshardingDonorRecipientCommonTest, ProcessReshardingFieldsWithoutDonorOr
     OperationContext* opCtx = operationContext();
     auto metadata =
         makeShardedMetadataForTemporaryReshardingCollection(opCtx, kThisShard.getShardId());
-    ScopedSetShardRole scopedSetShardRole{opCtx,
-                                          kTemporaryReshardingNss,
-                                          metadata.getShardVersion() /* shardVersion */,
-                                          boost::none /* databaseVersion */};
+    ScopedSetShardRole scopedSetShardRole{
+        opCtx,
+        kTemporaryReshardingNss,
+        ShardVersion(metadata.getShardVersion()) /* shardVersion */,
+        boost::none /* databaseVersion */};
 
     auto reshardingFields =
         createCommonReshardingFields(kReshardingUUID, CoordinatorStateEnum::kPreparingToDonate);
diff --git a/src/mongo/db/s/resharding/resharding_oplog_application.cpp b/src/mongo/db/s/resharding/resharding_oplog_application.cpp
index 21321ce00fd..ae4bc5f3d4e 100644
--- a/src/mongo/db/s/resharding/resharding_oplog_application.cpp
+++ b/src/mongo/db/s/resharding/resharding_oplog_application.cpp
@@ -81,7 +81,7 @@ void runWithTransaction(OperationContext* opCtx,
     // to allow the collection metadata information to be recovered.
     ScopedSetShardRole scopedSetShardRole(asr.opCtx(),
                                           nss,
-                                          ChunkVersion::IGNORED() /* shardVersion */,
+                                          ShardVersion::IGNORED() /* shardVersion */,
                                           boost::none /* databaseVersion */);
 
     MongoDOperationContextSession ocs(asr.opCtx());
diff --git a/src/mongo/db/s/resharding/resharding_oplog_batch_applier.cpp b/src/mongo/db/s/resharding/resharding_oplog_batch_applier.cpp
index f761376489d..fb5ddd671f6 100644
--- a/src/mongo/db/s/resharding/resharding_oplog_batch_applier.cpp
+++ b/src/mongo/db/s/resharding/resharding_oplog_batch_applier.cpp
@@ -90,7 +90,7 @@ SemiFuture<void> ReshardingOplogBatchApplier::applyBatch(
                            ScopedSetShardRole scopedSetShardRole(
                                opCtx.get(),
                                _crudApplication.getOutputNss(),
-                               ChunkVersion::IGNORED() /* shardVersion */,
+                               ShardVersion::IGNORED() /* shardVersion */,
                                boost::none /* databaseVersion */);
 
                            resharding::data_copy::withOneStaleConfigRetry(opCtx.get(), [&] {
diff --git a/src/mongo/db/s/sharding_write_router_bm.cpp b/src/mongo/db/s/sharding_write_router_bm.cpp
index 6d20ad82215..ce28ff604c5 100644
--- a/src/mongo/db/s/sharding_write_router_bm.cpp
+++ b/src/mongo/db/s/sharding_write_router_bm.cpp
@@ -150,7 +150,7 @@ std::unique_ptr<CatalogCacheMock> createCatalogCacheMock(OperationContext* opCtx
     OperationShardingState::setShardRole(
         opCtx,
         kNss,
-        chunkManager.getVersion(originatorShard) /* shardVersion */,
+        ShardVersion(chunkManager.getVersion(originatorShard)) /* shardVersion */,
         boost::none /* databaseVersion */);
 
     // Configuring the filtering metadata such that calls to getCollectionDescription return what we
diff --git a/src/mongo/db/service_entry_point_common.cpp b/src/mongo/db/service_entry_point_common.cpp
index a8b1f577776..22f79f800ef 100644
--- a/src/mongo/db/service_entry_point_common.cpp
+++ b/src/mongo/db/service_entry_point_common.cpp
@@ -1678,7 +1678,7 @@ void ExecCommandDatabase::_initiateCommand() {
             ? bucketNss
             : _invocation->ns();
 
-        boost::optional<ChunkVersion> shardVersion;
+        boost::optional<ShardVersion> shardVersion;
         if (auto shardVersionElem = request.body[ShardVersion::kShardVersionField]) {
             shardVersion = ShardVersion::parse(shardVersionElem);
         }
diff --git a/src/mongo/db/transaction/transaction_api_test.cpp b/src/mongo/db/transaction/transaction_api_test.cpp
index 1669995c946..7a2d5fb004b 100644
--- a/src/mongo/db/transaction/transaction_api_test.cpp
+++ b/src/mongo/db/transaction/transaction_api_test.cpp
@@ -1935,7 +1935,7 @@ TEST_F(TxnAPITest, MaxTimeMSIsSetIfOperationContextHasDeadlineAndIgnoresDefaultR
 
 TEST_F(TxnAPITest, CannotBeUsedWithinShardedOperationsIfClientDoesNotSupportIt) {
     OperationShardingState::setShardRole(
-        opCtx(), NamespaceString("foo.bar"), ChunkVersion(), boost::none);
+        opCtx(), NamespaceString("foo.bar"), ShardVersion(), boost::none);
 
     ASSERT_THROWS_CODE(
         resetTxnWithRetries(), DBException, ErrorCodes::duplicateCodeForTest(6638800));
@@ -1943,7 +1943,7 @@ TEST_F(TxnAPITest, CannotBeUsedWithinShardedOperationsIfClientDoesNotSupportIt)
 
 TEST_F(TxnAPITest, CanBeUsedWithinShardedOperationsIfClientSupportsIt) {
     OperationShardingState::setShardRole(
-        opCtx(), NamespaceString("foo.bar"), ChunkVersion(), boost::none);
+        opCtx(), NamespaceString("foo.bar"), ShardVersion(), boost::none);
 
     // Should not throw.
     resetTxnWithRetriesWithClient(std::make_unique<MockClusterOperationTransactionClient>());
diff --git a/src/mongo/db/ttl.cpp b/src/mongo/db/ttl.cpp
index d44ef72a6c6..ab31744bd54 100644
--- a/src/mongo/db/ttl.cpp
+++ b/src/mongo/db/ttl.cpp
@@ -368,7 +368,7 @@ bool TTLMonitor::_doTTLIndexDelete(OperationContext* opCtx,
         uassertStatusOK(userAllowedWriteNS(opCtx, *nss));
 
         // Attach IGNORED shard version to skip orphans (the range deleter will clear them up)
-        auto scopedRole = ScopedSetShardRole(opCtx, *nss, ChunkVersion::IGNORED(), boost::none);
+        auto scopedRole = ScopedSetShardRole(opCtx, *nss, ShardVersion::IGNORED(), boost::none);
         AutoGetCollection coll(opCtx, *nss, MODE_IX);
         // The collection with `uuid` might be renamed before the lock and the wrong namespace would
         // be locked and looked up so we double check here.
diff --git a/src/mongo/s/SConscript b/src/mongo/s/SConscript
index 98be1650966..07ee90c5051 100644
--- a/src/mongo/s/SConscript
+++ b/src/mongo/s/SConscript
@@ -152,6 +152,8 @@ env.Library(
         'chunk_version.idl',
         'database_version.cpp',
         'database_version.idl',
+        'index_version.cpp',
+        'index_version.idl',
         'mongod_and_mongos_server_parameters.idl',
         'request_types/abort_reshard_collection.idl',
         'request_types/add_shard_request_type.cpp',
@@ -189,6 +191,8 @@ env.Library(
         'shard_cannot_refresh_due_to_locks_held_exception.cpp',
         'shard_id.cpp',
         'shard_invalidated_for_targeting_exception.cpp',
+        'shard_version.cpp',
+        'shard_version.idl',
         'sharding_feature_flags.idl',
         'stale_exception.cpp',
         'type_collection_common_types.idl',
@@ -609,6 +613,7 @@ env.CppUnitTest(
         'sessions_collection_sharded_test.cpp',
         'shard_id_test.cpp',
         'shard_key_pattern_test.cpp',
+        'shard_version_test.cpp',
         'sharding_task_executor_test.cpp',
         'stale_exception_test.cpp',
         'stale_shard_version_helpers_test.cpp',
diff --git a/src/mongo/s/chunk_version.h b/src/mongo/s/chunk_version.h
index f573c406822..6bedb94f245 100644
--- a/src/mongo/s/chunk_version.h
+++ b/src/mongo/s/chunk_version.h
@@ -64,6 +64,17 @@ public:
     }
 
 protected:
+    static CollectionGeneration IGNORED() {
+        CollectionGeneration gen{OID(), Timestamp()};
+        gen._epoch.init(Date_t(), true);    // ignored OID is zero time, max machineId/inc
+        gen._timestamp = Timestamp::max();  // ignored Timestamp is the largest timestamp
+        return gen;
+    }
+
+    static CollectionGeneration UNSHARDED() {
+        return CollectionGeneration{OID(), Timestamp()};
+    }
+
     OID _epoch;
     Timestamp _timestamp;
 };
@@ -111,7 +122,7 @@ protected:
  * 3. (n, 0), n > 0 - invalid configuration.
  * 4. (n, m), n > 0, m > 0 - normal sharded collection version.
  */
-class ChunkVersion : public CollectionGeneration, public CollectionPlacement {
+class ChunkVersion : public virtual CollectionGeneration, public CollectionPlacement {
 public:
     /**
      * The name for the chunk version information field, which ddl operations use to send only
@@ -128,17 +139,14 @@ public:
      * Indicates that the collection is not sharded.
      */
     static ChunkVersion UNSHARDED() {
-        return ChunkVersion();
+        return ChunkVersion(CollectionGeneration::UNSHARDED(), {0, 0});
     }
 
     /**
      * Indicates that the shard version checking must be skipped.
      */
     static ChunkVersion IGNORED() {
-        ChunkVersion version;
-        version._epoch.init(Date_t(), true);    // ignored OID is zero time, max machineId/inc
-        version._timestamp = Timestamp::max();  // ignored Timestamp is the largest timestamp
-        return version;
+        return ChunkVersion(CollectionGeneration::IGNORED(), {0, 0});
     }
 
     static bool isIgnoredVersion(const ChunkVersion& version) {
diff --git a/src/mongo/s/commands/cluster_split_cmd.cpp b/src/mongo/s/commands/cluster_split_cmd.cpp
index b67bfc12732..6bfabdb5b76 100644
--- a/src/mongo/s/commands/cluster_split_cmd.cpp
+++ b/src/mongo/s/commands/cluster_split_cmd.cpp
@@ -59,14 +59,14 @@ BSONObj selectMedianKey(OperationContext* opCtx,
                         const ShardId& shardId,
                         const NamespaceString& nss,
                         const ShardKeyPattern& shardKeyPattern,
-                        const ChunkVersion& chunkVersion,
+                        const ShardVersion& chunkVersion,
                         const ChunkRange& chunkRange) {
     BSONObjBuilder cmd;
     cmd.append("splitVector", nss.ns());
     cmd.append("keyPattern", shardKeyPattern.toBSON());
     chunkRange.append(&cmd);
     cmd.appendBool("force", true);
-    chunkVersion.serialize(ChunkVersion::kChunkVersionField, &cmd);
+    chunkVersion.serialize(ShardVersion::kShardVersionField, &cmd);
 
     auto shard = uassertStatusOK(Grid::get(opCtx)->shardRegistry()->getShard(opCtx, shardId));
 
diff --git a/src/mongo/s/index_version.cpp b/src/mongo/s/index_version.cpp
new file mode 100644
index 00000000000..38cca4cee7e
--- /dev/null
+++ b/src/mongo/s/index_version.cpp
@@ -0,0 +1,37 @@
+/**
+ *    Copyright (C) 2022-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/s/index_version.h"
+namespace mongo {
+
+std::string CollectionIndexes::toString() const {
+    return _indexVersion ? _indexVersion->toString() : "";
+}
+
+}  // namespace mongo
diff --git a/src/mongo/s/index_version.h b/src/mongo/s/index_version.h
new file mode 100644
index 00000000000..47756040008
--- /dev/null
+++ b/src/mongo/s/index_version.h
@@ -0,0 +1,64 @@
+/**
+ *    Copyright (C) 2022-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include "mongo/s/chunk_version.h"
+
+namespace mongo {
+
+/**
+ * Reflects the index information about a collection.
+ */
+class CollectionIndexes : public virtual CollectionGeneration {
+public:
+    CollectionIndexes(CollectionGeneration generation, boost::optional<Timestamp> index)
+        : CollectionGeneration(generation), _indexVersion(index) {}
+
+    CollectionIndexes() : CollectionIndexes({OID(), Timestamp()}, {Timestamp()}) {}
+
+    static CollectionIndexes IGNORED() {
+        return CollectionIndexes(CollectionGeneration::IGNORED(), {Timestamp()});
+    }
+
+    static CollectionIndexes UNSHARDED() {
+        return CollectionIndexes(CollectionGeneration::UNSHARDED(), {Timestamp()});
+    }
+
+    boost::optional<Timestamp> indexVersion() const {
+        return _indexVersion;
+    }
+
+    std::string toString() const;
+
+protected:
+    boost::optional<Timestamp> _indexVersion;
+};
+
+}  // namespace mongo
diff --git a/src/mongo/s/index_version.idl b/src/mongo/s/index_version.idl
new file mode 100644
index 00000000000..74d62224946
--- /dev/null
+++ b/src/mongo/s/index_version.idl
@@ -0,0 +1,45 @@
+# Copyright (C) 2022-present MongoDB, Inc.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the Server Side Public License, version 1,
+# as published by MongoDB, Inc.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# Server Side Public License for more details.
+#
+# You should have received a copy of the Server Side Public License
+# along with this program. If not, see
+# <http://www.mongodb.com/licensing/server-side-public-license>.
+#
+# As a special exception, the copyright holders give permission to link the
+# code of portions of this program with the OpenSSL library under certain
+# conditions as described in each individual source file and distribute
+# linked combinations including the program with the OpenSSL library. You
+# must comply with the Server Side Public License in all respects for
+# all of the code used other than as permitted herein. If you modify file(s)
+# with this exception, you may extend this exception to your version of the
+# file(s), but you are not obligated to do so. If you do not wish to do so,
+# delete this exception statement from your version. If you delete this
+# exception statement from all source files in the program, then also delete
+# it in the license file.
+#
+
+global:
+    cpp_namespace: "mongo"
+
+imports:
+    - "mongo/idl/basic_types.idl"
+
+structs:
+    CollectionIndexesBase:
+        description: Represents the index part of a collection version, which is subordinate to the
+                      collection generation.
+        strict: false
+        fields:
+            i:
+                type: timestamp
+                description: "Collection index version"
+                cpp_name: IndexVersion
+                optional: true
diff --git a/src/mongo/s/query/cluster_find.cpp b/src/mongo/s/query/cluster_find.cpp
index 187ee7289c0..bb55dc6ccc6 100644
--- a/src/mongo/s/query/cluster_find.cpp
+++ b/src/mongo/s/query/cluster_find.cpp
@@ -189,8 +189,7 @@ std::vector<std::pair<ShardId, BSONObj>> constructRequestsForShards(
             ShardVersion(cm.getVersion(shardId))
                 .serialize(ShardVersion::kShardVersionField, &cmdBuilder);
         } else if (!query.nss().isOnInternalDb()) {
-            ShardVersion(ChunkVersion::UNSHARDED())
-                .serialize(ShardVersion::kShardVersionField, &cmdBuilder);
+            ShardVersion::UNSHARDED().serialize(ShardVersion::kShardVersionField, &cmdBuilder);
             cmdBuilder.append("databaseVersion", cm.dbVersion().toBSON());
         }
 
diff --git a/src/mongo/s/router.cpp b/src/mongo/s/router.cpp
index 5149289cb43..ed4b474e8f1 100644
--- a/src/mongo/s/router.cpp
+++ b/src/mongo/s/router.cpp
@@ -62,7 +62,7 @@ void DBPrimaryRouter::appendCRUDUnshardedRoutingTokenToCommand(const ShardId& sh
         BSONObjBuilder dbvBuilder(builder->subobjStart(DatabaseVersion::kDatabaseVersionField));
         dbVersion.serialize(&dbvBuilder);
     }
-    ShardVersion(ChunkVersion::UNSHARDED()).serialize(ShardVersion::kShardVersionField, builder);
+    ShardVersion::UNSHARDED().serialize(ShardVersion::kShardVersionField, builder);
 }
 
 CachedDatabaseInfo DBPrimaryRouter::_getRoutingInfo(OperationContext* opCtx) const {
diff --git a/src/mongo/s/shard_version.cpp b/src/mongo/s/shard_version.cpp
new file mode 100644
index 00000000000..426a33d1428
--- /dev/null
+++ b/src/mongo/s/shard_version.cpp
@@ -0,0 +1,69 @@
+/**
+ *    Copyright (C) 2022-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/s/shard_version.h"
+
+#include "mongo/s/shard_version_gen.h"
+
+namespace mongo {
+
+ShardVersion::ShardVersion(ChunkVersion chunkVersion, CollectionIndexes indexVersion)
+    : CollectionGeneration([&]() {
+          uassert(ErrorCodes::BadValue,
+                  "ChunkVersion and CollectionIndexes have different generations",
+                  chunkVersion.isSameCollection(indexVersion));
+          return CollectionGeneration(chunkVersion.epoch(), chunkVersion.getTimestamp());
+      }()),
+      ChunkVersion(chunkVersion),
+      CollectionIndexes(indexVersion) {}
+
+ShardVersion ShardVersion::parse(const BSONElement& element) {
+    auto parsedVersion = ShardVersionBase::parse(IDLParserContext("ShardVersion"), element.Obj());
+    auto version = parsedVersion.getVersion();
+    return ShardVersion(ChunkVersion({parsedVersion.getEpoch(), parsedVersion.getTimestamp()},
+                                     {version.getSecs(), version.getInc()}),
+                        CollectionIndexes({parsedVersion.getEpoch(), parsedVersion.getTimestamp()},
+                                          {parsedVersion.getIndexVersion()}));
+}
+
+void ShardVersion::serialize(StringData field, BSONObjBuilder* builder) const {
+    ShardVersionBase version;
+    version.setGeneration({_epoch, _timestamp});
+    version.setPlacement(Timestamp(majorVersion(), minorVersion()));
+    CollectionIndexesBase indexVersion;
+    indexVersion.setIndexVersion(_indexVersion);
+    version.setIndex(indexVersion);
+    builder->append(field, version.toBSON());
+}
+
+std::string ShardVersion::toString() const {
+    return CollectionIndexes(*this).toString() + "||" + ChunkVersion(*this).toString();
+}
+
+}  // namespace mongo
diff --git a/src/mongo/s/shard_version.h b/src/mongo/s/shard_version.h
index f49cb10f83a..2c48fd9a976 100644
--- a/src/mongo/s/shard_version.h
+++ b/src/mongo/s/shard_version.h
@@ -28,7 +28,8 @@
  */
 #pragma once
 
-#include "mongo/s/chunk_version_gen.h"
+#include "mongo/s/chunk_version.h"
+#include "mongo/s/index_version.h"
 
 namespace mongo {
 
@@ -39,7 +40,7 @@ namespace mongo {
  * network requests and the shard versioning protocol.
  *
  */
-class ShardVersion : public ChunkVersion {
+class ShardVersion : public ChunkVersion, public CollectionIndexes {
 public:
     /**
      * The name for the shard version information field, which shard-aware commands should include
@@ -47,24 +48,27 @@ public:
      */
     static constexpr StringData kShardVersionField = "shardVersion"_sd;
 
-    ShardVersion(ChunkVersion chunkVersion) : ChunkVersion(chunkVersion) {}
+    ShardVersion(ChunkVersion chunkVersion, CollectionIndexes indexVersion);
 
-    ShardVersion() : ShardVersion(ChunkVersion()) {}
+    ShardVersion(ChunkVersion chunkVersion)
+        : CollectionGeneration(chunkVersion.epoch(), chunkVersion.getTimestamp()),
+          ChunkVersion(chunkVersion),
+          CollectionIndexes() {}
 
-    static ShardVersion parse(const BSONElement& element) {
-        auto parsedVersion =
-            ChunkVersion60Format::parse(IDLParserContext("ShardVersion"), element.Obj());
-        auto version = parsedVersion.getVersion();
-        return ShardVersion(ChunkVersion({parsedVersion.getEpoch(), parsedVersion.getTimestamp()},
-                                         {version.getSecs(), version.getInc()}));
+    ShardVersion() : ShardVersion(ChunkVersion(), CollectionIndexes()) {}
+
+    static ShardVersion IGNORED() {
+        return ShardVersion(ChunkVersion::IGNORED(), CollectionIndexes::IGNORED());
     }
 
-    void serialize(StringData field, BSONObjBuilder* builder) const {
-        ChunkVersion60Format version;
-        version.setGeneration({_epoch, _timestamp});
-        version.setPlacement(Timestamp(majorVersion(), minorVersion()));
-        builder->append(field, version.toBSON());
+    static ShardVersion UNSHARDED() {
+        return ShardVersion(ChunkVersion::UNSHARDED(), CollectionIndexes::UNSHARDED());
     }
+
+    static ShardVersion parse(const BSONElement& element);
+    void serialize(StringData field, BSONObjBuilder* builder) const;
+
+    std::string toString() const;
 };
 
 }  // namespace mongo
diff --git a/src/mongo/s/shard_version.idl b/src/mongo/s/shard_version.idl
new file mode 100644
index 00000000000..9d584cca200
--- /dev/null
+++ b/src/mongo/s/shard_version.idl
@@ -0,0 +1,46 @@
+# Copyright (C) 2022-present MongoDB, Inc.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the Server Side Public License, version 1,
+# as published by MongoDB, Inc.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# Server Side Public License for more details.
+#
+# You should have received a copy of the Server Side Public License
+# along with this program. If not, see
+# <http://www.mongodb.com/licensing/server-side-public-license>.
+#
+# As a special exception, the copyright holders give permission to link the
+# code of portions of this program with the OpenSSL library under certain
+# conditions as described in each individual source file and distribute
+# linked combinations including the program with the OpenSSL library. You
+# must comply with the Server Side Public License in all respects for
+# all of the code used other than as permitted herein. If you modify file(s)
+# with this exception, you may extend this exception to your version of the
+# file(s), but you are not obligated to do so. If you do not wish to do so,
+# delete this exception statement from your version. If you delete this
+# exception statement from all source files in the program, then also delete
+# it in the license file.
+#
+
+global:
+    cpp_namespace: "mongo"
+    cpp_includes:
+        - "mongo/s/shard_version.h"
+
+imports:
+    - "mongo/s/chunk_version.idl"
+    - "mongo/s/index_version.idl"
+
+structs:
+    ShardVersionBase:
+        description: Shard version for the shard version protocol. Includes collection generation,
+                     placement version, and index version.
+        strict: false
+        chained_structs:
+            CollectionGenerationBase: Generation
+            CollectionPlacementBase: Placement
+            CollectionIndexesBase: Index
diff --git a/src/mongo/s/shard_version_test.cpp b/src/mongo/s/shard_version_test.cpp
new file mode 100644
index 00000000000..9bd7858d1ec
--- /dev/null
+++ b/src/mongo/s/shard_version_test.cpp
@@ -0,0 +1,70 @@
+/**
+ *    Copyright (C) 2022-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+#include "mongo/s/shard_version.h"
+#include "mongo/unittest/unittest.h"
+
+namespace mongo {
+namespace {
+
+TEST(ShardVersionTest, ConstructWithDifferentGenerationThrows) {
+    const CollectionGeneration gen1(OID::gen(), Timestamp(1, 2));
+    const CollectionGeneration gen2(OID::gen(), Timestamp(2, 1));
+    const ChunkVersion chunkVersion(gen1, {3, 4});
+    const CollectionIndexes collectionIndexes(gen2, {Timestamp(5, 6)});
+    ASSERT_THROWS_CODE(
+        ShardVersion(chunkVersion, collectionIndexes), DBException, ErrorCodes::BadValue);
+}
+
+TEST(ShardVersionTest, ConstructCorrectly) {
+    const CollectionGeneration gen(OID::gen(), Timestamp(1, 2));
+    const ChunkVersion chunkVersion(gen, {3, 4});
+    const CollectionIndexes collectionIndexes(gen, {Timestamp(5, 6)});
+    const ShardVersion shardVersion(chunkVersion, collectionIndexes);
+    ASSERT_EQ(shardVersion.getTimestamp(), Timestamp(1, 2));
+    ASSERT_EQ(shardVersion.majorVersion(), 3);
+    ASSERT_EQ(shardVersion.minorVersion(), 4);
+    ASSERT_EQ(shardVersion.indexVersion(), Timestamp(5, 6));
+}
+
+TEST(ShardVersionTest, ToAndFromBSON) {
+    const CollectionGeneration gen(OID::gen(), Timestamp(1, 2));
+    const ChunkVersion chunkVersion(gen, {3, 4});
+    const CollectionIndexes collectionIndexes(gen, {Timestamp(5, 6)});
+    const ShardVersion shardVersion(chunkVersion, collectionIndexes);
+
+    BSONObjBuilder builder;
+    shardVersion.serialize(ShardVersion::kShardVersionField, &builder);
+    const auto obj = builder.obj();
+
+    const auto fromBSON = ShardVersion::parse(obj[ShardVersion::kShardVersionField]);
+    ASSERT_EQ(fromBSON, shardVersion);
+}
+
+}  // namespace
+}  // namespace mongo