diff --git a/ent/src/yb/master/catalog_manager.h b/ent/src/yb/master/catalog_manager.h
index 9c1570ebc5c3..ca64bef8668e 100644
--- a/ent/src/yb/master/catalog_manager.h
+++ b/ent/src/yb/master/catalog_manager.h
@@ -242,7 +242,8 @@ class CatalogManager : public yb::master::CatalogManager, SnapshotCoordinatorCon
   Result<SysRowEntries> CollectEntries(
       const google::protobuf::RepeatedPtrField<TableIdentifierPB>& tables,
       bool add_indexes,
-      bool include_parent_colocated_table) override;
+      bool include_parent_colocated_table,
+      bool succeed_if_create_in_progress) override;
 
   server::Clock* Clock() override;
 
@@ -269,6 +270,8 @@ class CatalogManager : public yb::master::CatalogManager, SnapshotCoordinatorCon
 
   bool IsLeader() override;
 
+  Result<SnapshotSchedulesToTabletsMap> MakeSnapshotSchedulesToTabletsMap() override;
+
   static void SetTabletSnapshotsState(SysSnapshotEntryPB::State state,
                                       SysSnapshotEntryPB* snapshot_pb);
 
diff --git a/ent/src/yb/master/catalog_manager_ent.cc b/ent/src/yb/master/catalog_manager_ent.cc
index d74996e96c72..8cc20eccee82 100644
--- a/ent/src/yb/master/catalog_manager_ent.cc
+++ b/ent/src/yb/master/catalog_manager_ent.cc
@@ -383,10 +383,12 @@ void CatalogManager::Submit(std::unique_ptr<tablet::Operation> operation) {
 Result<SysRowEntries> CatalogManager::CollectEntries(
     const google::protobuf::RepeatedPtrField<TableIdentifierPB>& table_identifiers,
     bool add_indexes,
-    bool include_parent_colocated_table) {
+    bool include_parent_colocated_table,
+    bool succeed_if_create_in_progress) {
   SysRowEntries entries;
   auto tables = VERIFY_RESULT(CollectTables(
-      table_identifiers, add_indexes, include_parent_colocated_table));
+      table_identifiers, add_indexes, include_parent_colocated_table,
+      succeed_if_create_in_progress));
   for (const auto& table : tables) {
     // TODO(txn_snapshot) use single lock to resolve all tables to tablets
     SnapshotInfo::AddEntries(table, entries.mutable_entries(), /* tablet_infos= */ nullptr);
@@ -402,7 +404,7 @@ server::Clock* CatalogManager::Clock() {
 Status CatalogManager::CreateTransactionAwareSnapshot(
     const CreateSnapshotRequestPB& req, CreateSnapshotResponsePB* resp, rpc::RpcContext* rpc) {
   SysRowEntries entries = VERIFY_RESULT(CollectEntries(
-      req.tables(), req.add_indexes(), true /* include_parent_colocated_table */));
+      req.tables(), req.add_indexes(), true /* include_parent_colocated_table */, false));
 
   auto snapshot_id = VERIFY_RESULT(snapshot_coordinator_.Create(
       entries, req.imported(), rpc->GetClientDeadline()));
@@ -3333,6 +3335,10 @@ void CatalogManager::Started() {
   snapshot_coordinator_.Start();
 }
 
+Result<SnapshotSchedulesToTabletsMap> CatalogManager::MakeSnapshotSchedulesToTabletsMap() {
+  return snapshot_coordinator_.MakeSnapshotSchedulesToTabletsMap();
+}
+
 } // namespace enterprise
 }  // namespace master
 }  // namespace yb
diff --git a/src/yb/client/snapshot-schedule-test.cc b/src/yb/client/snapshot-schedule-test.cc
index 846db10d0126..1af484f0bf07 100644
--- a/src/yb/client/snapshot-schedule-test.cc
+++ b/src/yb/client/snapshot-schedule-test.cc
@@ -13,11 +13,16 @@
 
 #include "yb/client/snapshot_test_base.h"
 
+#include "yb/client/session.h"
+
+#include "yb/master/master.h"
 #include "yb/master/master_backup.proxy.h"
 
 #include "yb/tablet/tablet_peer.h"
 #include "yb/tablet/tablet_retention_policy.h"
 
+#include "yb/yql/cql/ql/util/statement_result.h"
+
 using namespace std::literals;
 
 DECLARE_bool(enable_history_cutoff_propagation);
@@ -183,6 +188,55 @@ TEST_F(SnapshotScheduleTest, GC) {
   }
 }
 
+TEST_F(SnapshotScheduleTest, Index) {
+  FLAGS_timestamp_history_retention_interval_sec = kTimeMultiplier;
+
+  auto schedule_id = ASSERT_RESULT(CreateSchedule());
+  ASSERT_OK(WaitScheduleSnapshot(schedule_id));
+
+  CreateIndex(Transactional::kTrue, 1, false);
+  auto hybrid_time = cluster_->mini_master(0)->master()->clock()->Now();
+  constexpr int kTransaction = 0;
+  constexpr auto op_type = WriteOpType::INSERT;
+
+  auto session = CreateSession();
+  for (size_t r = 0; r != kNumRows; ++r) {
+    ASSERT_OK(kv_table_test::WriteRow(
+        &index_, session, KeyForTransactionAndIndex(kTransaction, r),
+        ValueForTransactionAndIndex(kTransaction, r, op_type), op_type));
+  }
+
+  LOG(INFO) << "Index columns: " << AsString(index_.AllColumnNames());
+  for (size_t r = 0; r != kNumRows; ++r) {
+    const auto key = KeyForTransactionAndIndex(kTransaction, r);
+    const auto fetched = ASSERT_RESULT(kv_table_test::SelectRow(
+        &index_, session, key, kValueColumn));
+    ASSERT_EQ(key, fetched);
+  }
+
+  auto peers = ListTabletPeers(cluster_.get(), [index_id = index_->id()](const auto& peer) {
+    return peer->tablet_metadata()->table_id() == index_id;
+  });
+
+  ASSERT_OK(WaitFor([this, peers, hybrid_time]() -> Result<bool> {
+    auto snapshots = VERIFY_RESULT(ListSnapshots());
+    if (snapshots.size() == 2) {
+      return true;
+    }
+
+    for (const auto& peer : peers) {
+      SCOPED_TRACE(Format(
+          "T $0 P $1 Table $2", peer->tablet_id(), peer->permanent_uuid(),
+          peer->tablet_metadata()->table_name()));
+      auto tablet = peer->tablet();
+      auto history_cutoff = tablet->RetentionPolicy()->GetRetentionDirective().history_cutoff;
+      SCHECK_LE(history_cutoff, hybrid_time, IllegalState, "Too big history cutoff");
+    }
+
+    return false;
+  }, kSnapshotInterval, "Second snapshot"));
+}
+
 TEST_F(SnapshotScheduleTest, Restart) {
   ASSERT_NO_FATALS(WriteData());
   auto schedule_id = ASSERT_RESULT(CreateSchedule());
diff --git a/src/yb/common/snapshot.h b/src/yb/common/snapshot.h
index f7e6a5300df2..0901a0915ee4 100644
--- a/src/yb/common/snapshot.h
+++ b/src/yb/common/snapshot.h
@@ -14,6 +14,10 @@
 #ifndef YB_COMMON_SNAPSHOT_H
 #define YB_COMMON_SNAPSHOT_H
 
+#include <unordered_map>
+
+#include "yb/common/entity_ids.h"
+
 #include "yb/util/strongly_typed_uuid.h"
 
 namespace yb {
@@ -22,6 +26,9 @@ YB_STRONGLY_TYPED_UUID(TxnSnapshotId);
 YB_STRONGLY_TYPED_UUID(TxnSnapshotRestorationId);
 YB_STRONGLY_TYPED_UUID(SnapshotScheduleId);
 
+using SnapshotSchedulesToTabletsMap =
+    std::unordered_map<SnapshotScheduleId, std::vector<TabletId>, SnapshotScheduleIdHash>;
+
 } // namespace yb
 
 #endif // YB_COMMON_SNAPSHOT_H
diff --git a/src/yb/master/async_rpc_tasks.cc b/src/yb/master/async_rpc_tasks.cc
index aa79c6f54999..66588ef98a33 100644
--- a/src/yb/master/async_rpc_tasks.cc
+++ b/src/yb/master/async_rpc_tasks.cc
@@ -479,7 +479,8 @@ TabletServerId AsyncTabletLeaderTask::permanent_uuid() const {
 AsyncCreateReplica::AsyncCreateReplica(Master *master,
                                        ThreadPool *callback_pool,
                                        const string& permanent_uuid,
-                                       const scoped_refptr<TabletInfo>& tablet)
+                                       const scoped_refptr<TabletInfo>& tablet,
+                                       const std::vector<SnapshotScheduleId>& snapshot_schedules)
   : RetrySpecificTSRpcTask(master, callback_pool, permanent_uuid, tablet->table().get()),
     tablet_id_(tablet->tablet_id()) {
   deadline_ = start_ts_;
@@ -503,6 +504,11 @@ AsyncCreateReplica::AsyncCreateReplica(Master *master,
   if (table_lock->data().pb.has_index_info()) {
     req_.mutable_index_info()->CopyFrom(table_lock->data().pb.index_info());
   }
+  auto& req_schedules = *req_.mutable_snapshot_schedules();
+  req_schedules.Reserve(snapshot_schedules.size());
+  for (const auto& id : snapshot_schedules) {
+    req_schedules.Add()->assign(id.AsSlice().cdata(), id.size());
+  }
 }
 
 void AsyncCreateReplica::HandleResponse(int attempt) {
diff --git a/src/yb/master/async_rpc_tasks.h b/src/yb/master/async_rpc_tasks.h
index b130de59ebee..f54bf5940fec 100644
--- a/src/yb/master/async_rpc_tasks.h
+++ b/src/yb/master/async_rpc_tasks.h
@@ -296,7 +296,8 @@ class AsyncCreateReplica : public RetrySpecificTSRpcTask {
   AsyncCreateReplica(Master *master,
                      ThreadPool *callback_pool,
                      const std::string& permanent_uuid,
-                     const scoped_refptr<TabletInfo>& tablet);
+                     const scoped_refptr<TabletInfo>& tablet,
+                     const std::vector<SnapshotScheduleId>& snapshot_schedules);
 
   Type type() const override { return ASYNC_CREATE_REPLICA; }
 
diff --git a/src/yb/master/catalog_manager.cc b/src/yb/master/catalog_manager.cc
index 112788b3fad6..c96aaaf967e0 100644
--- a/src/yb/master/catalog_manager.cc
+++ b/src/yb/master/catalog_manager.cc
@@ -3553,7 +3553,8 @@ Status CatalogManager::FindNamespace(const NamespaceIdentifierPB& ns_identifier,
   return FindNamespaceUnlocked(ns_identifier, ns_info);
 }
 
-Result<TableDescription> CatalogManager::DescribeTable(const TableIdentifierPB& table_identifier) {
+Result<TableDescription> CatalogManager::DescribeTable(
+    const TableIdentifierPB& table_identifier, bool succeed_if_create_in_progress) {
   TableDescription result;
 
   // Lookup the table and verify it exists.
@@ -3569,7 +3570,7 @@ Result<TableDescription> CatalogManager::DescribeTable(const TableIdentifierPB&
     TRACE("Locking table");
     auto l = result.table_info->LockForRead();
 
-    if (result.table_info->IsCreateInProgress()) {
+    if (!succeed_if_create_in_progress && result.table_info->IsCreateInProgress()) {
       return STATUS(IllegalState, "Table creation is in progress", result.table_info->ToString(),
                     MasterError(MasterErrorPB::TABLE_CREATION_IS_IN_PROGRESS));
     }
@@ -7842,8 +7843,7 @@ Status CatalogManager::ProcessPendingAssignments(const TabletInfos& tablets) {
     }
   }
   // Send the CreateTablet() requests to the servers. This is asynchronous / non-blocking.
-  SendCreateTabletRequests(deferred.needs_create_rpc);
-  return Status::OK();
+  return SendCreateTabletRequests(deferred.needs_create_rpc);
 }
 
 Status CatalogManager::SelectReplicasForTablet(const TSDescriptorVector& ts_descs,
@@ -7978,18 +7978,27 @@ Status CatalogManager::HandlePlacementUsingPlacementInfo(const PlacementInfoPB&
   return Status::OK();
 }
 
-void CatalogManager::SendCreateTabletRequests(const vector<TabletInfo*>& tablets) {
+Status CatalogManager::SendCreateTabletRequests(const vector<TabletInfo*>& tablets) {
+  auto schedules_to_tablets_map = VERIFY_RESULT(MakeSnapshotSchedulesToTabletsMap());
   for (TabletInfo *tablet : tablets) {
     const consensus::RaftConfigPB& config =
         tablet->metadata().dirty().pb.committed_consensus_state().config();
     tablet->set_last_update_time(MonoTime::Now());
+    std::vector<SnapshotScheduleId> schedules;
+    for (const auto& pair : schedules_to_tablets_map) {
+      if (std::binary_search(pair.second.begin(), pair.second.end(), tablet->id())) {
+        schedules.push_back(pair.first);
+      }
+    }
     for (const RaftPeerPB& peer : config.peers()) {
       auto task = std::make_shared<AsyncCreateReplica>(master_, AsyncTaskPool(),
-          peer.permanent_uuid(), tablet);
+          peer.permanent_uuid(), tablet, schedules);
       tablet->table()->AddTask(task);
       WARN_NOT_OK(ScheduleTask(task), "Failed to send new tablet request");
     }
   }
+
+  return Status::OK();
 }
 
 // If responses have been received from sufficient replicas (including hinted leader),
@@ -9003,12 +9012,14 @@ Status CatalogManager::ScheduleTask(std::shared_ptr<RetryingTSRpcTask> task) {
 Result<vector<TableDescription>> CatalogManager::CollectTables(
     const google::protobuf::RepeatedPtrField<TableIdentifierPB>& tables,
     bool add_indexes,
-    bool include_parent_colocated_table) {
+    bool include_parent_colocated_table,
+    bool succeed_if_create_in_progress) {
   vector<TableDescription> all_tables;
   unordered_set<NamespaceId> parent_colocated_table_ids;
 
   for (const auto& table_id_pb : tables) {
-    TableDescription table_description = VERIFY_RESULT(DescribeTable(table_id_pb));
+    auto table_description = VERIFY_RESULT(DescribeTable(
+        table_id_pb, succeed_if_create_in_progress));
     if (include_parent_colocated_table && table_description.table_info->colocated()) {
       // If a table is colocated, add its parent colocated table as well.
       const auto parent_table_id =
@@ -9019,8 +9030,8 @@ Result<vector<TableDescription>> CatalogManager::CollectTables(
         TableIdentifierPB parent_table_pb;
         parent_table_pb.set_table_id(parent_table_id);
         parent_table_pb.mutable_namespace_()->set_id(table_description.namespace_info->id());
-        TableDescription parent_table_description = VERIFY_RESULT(DescribeTable(parent_table_pb));
-        all_tables.push_back(parent_table_description);
+        all_tables.push_back(VERIFY_RESULT(DescribeTable(
+            parent_table_pb, succeed_if_create_in_progress)));
       }
     }
     all_tables.push_back(table_description);
@@ -9047,7 +9058,8 @@ Result<vector<TableDescription>> CatalogManager::CollectTables(
         TableIdentifierPB index_id_pb;
         index_id_pb.set_table_id(index_info.table_id());
         index_id_pb.mutable_namespace_()->set_id(table_description.namespace_info->id());
-        all_tables.push_back(VERIFY_RESULT(DescribeTable(index_id_pb)));
+        all_tables.push_back(VERIFY_RESULT(DescribeTable(
+            index_id_pb, succeed_if_create_in_progress)));
       }
     }
   }
diff --git a/src/yb/master/catalog_manager.h b/src/yb/master/catalog_manager.h
index 311ce4a95f32..955841ec0271 100644
--- a/src/yb/master/catalog_manager.h
+++ b/src/yb/master/catalog_manager.h
@@ -689,7 +689,8 @@ class CatalogManager : public tserver::TabletPeerLookupIf {
   CHECKED_STATUS FindTable(const TableIdentifierPB& table_identifier,
                            scoped_refptr<TableInfo>* table_info);
 
-  Result<TableDescription> DescribeTable(const TableIdentifierPB& table_identifier);
+  Result<TableDescription> DescribeTable(
+      const TableIdentifierPB& table_identifier, bool succeed_if_create_in_progress);
 
   void AssertLeaderLockAcquiredForReading() const {
     leader_lock_.AssertAcquiredForReading();
@@ -742,7 +743,8 @@ class CatalogManager : public tserver::TabletPeerLookupIf {
   Result<std::vector<TableDescription>> CollectTables(
       const google::protobuf::RepeatedPtrField<TableIdentifierPB>& tables,
       bool add_indexes,
-      bool include_parent_colocated_table = false);
+      bool include_parent_colocated_table = false,
+      bool succeed_if_create_in_progress = false);
 
   // Returns 'table_replication_info' itself if set. Else looks up placement info for its
   // 'tablespace_id'. If neither is set, returns the cluster level replication info.
@@ -1023,7 +1025,7 @@ class CatalogManager : public tserver::TabletPeerLookupIf {
   //
   // This must be called after persisting the tablet state as
   // CREATING to ensure coherent state after Master failover.
-  void SendCreateTabletRequests(const std::vector<TabletInfo*>& tablets);
+  CHECKED_STATUS SendCreateTabletRequests(const std::vector<TabletInfo*>& tablets);
 
   // Send the "alter table request" to all tablets of the specified table.
   //
@@ -1233,6 +1235,10 @@ class CatalogManager : public tserver::TabletPeerLookupIf {
   // the cluster config affinity specification.
   CHECKED_STATUS SysCatalogRespectLeaderAffinity();
 
+  virtual Result<SnapshotSchedulesToTabletsMap> MakeSnapshotSchedulesToTabletsMap() {
+    return SnapshotSchedulesToTabletsMap();
+  }
+
   // ----------------------------------------------------------------------------------------------
   // Private member fields
   // ----------------------------------------------------------------------------------------------
diff --git a/src/yb/master/master_snapshot_coordinator.cc b/src/yb/master/master_snapshot_coordinator.cc
index f6c462ff1302..338cfec1bb4d 100644
--- a/src/yb/master/master_snapshot_coordinator.cc
+++ b/src/yb/master/master_snapshot_coordinator.cc
@@ -384,6 +384,28 @@ class MasterSnapshotCoordinator::Impl {
     return Status::OK();
   }
 
+  Result<SnapshotSchedulesToTabletsMap> MakeSnapshotSchedulesToTabletsMap() {
+    std::vector<std::pair<SnapshotScheduleId, SnapshotScheduleFilterPB>> schedules;
+    {
+      std::lock_guard<std::mutex> lock(mutex_);
+      for (const auto& schedule : schedules_) {
+        schedules.emplace_back(schedule->id(), schedule->options().filter());
+      }
+    }
+    SnapshotSchedulesToTabletsMap result;
+    for (const auto& id_and_filter : schedules) {
+      auto entries = VERIFY_RESULT(CollectEntries(id_and_filter.second));
+      auto& tablets = result[id_and_filter.first];
+      for (const auto& entry : entries.entries()) {
+        if (entry.type() == SysRowEntry::TABLET) {
+          tablets.push_back(entry.id());
+        }
+      }
+      std::sort(tablets.begin(), tablets.end());
+    }
+    return result;
+  }
+
   void Start() {
     poller_.Start(&context_.Scheduler(), FLAGS_snapshot_coordinator_poll_interval_ms * 1ms);
   }
@@ -601,8 +623,7 @@ class MasterSnapshotCoordinator::Impl {
   }
 
   CHECKED_STATUS ExecuteScheduleOperation(const SnapshotScheduleOperation& operation) {
-    auto entries = VERIFY_RESULT(context_.CollectEntries(
-        operation.filter.tables().tables(), true, true));
+    auto entries = VERIFY_RESULT(CollectEntries(operation.filter));
     RETURN_NOT_OK(SubmitCreate(
         entries, false, operation.schedule_id, operation.snapshot_id,
         tablet::MakeFunctorOperationCompletionCallback(
@@ -757,6 +778,10 @@ class MasterSnapshotCoordinator::Impl {
     return Status::OK();
   }
 
+  Result<SysRowEntries> CollectEntries(const SnapshotScheduleFilterPB& filter) {
+    return context_.CollectEntries(filter.tables().tables(), true, true, true);
+  }
+
   SnapshotCoordinatorContext& context_;
   std::mutex mutex_;
   class ScheduleTag;
@@ -880,5 +905,10 @@ Status MasterSnapshotCoordinator::FillHeartbeatResponse(TSHeartbeatResponsePB* r
   return impl_->FillHeartbeatResponse(resp);
 }
 
+Result<SnapshotSchedulesToTabletsMap>
+    MasterSnapshotCoordinator::MakeSnapshotSchedulesToTabletsMap() {
+  return impl_->MakeSnapshotSchedulesToTabletsMap();
+}
+
 } // namespace master
 } // namespace yb
diff --git a/src/yb/master/master_snapshot_coordinator.h b/src/yb/master/master_snapshot_coordinator.h
index a578cfabf27d..6c86e88248ea 100644
--- a/src/yb/master/master_snapshot_coordinator.h
+++ b/src/yb/master/master_snapshot_coordinator.h
@@ -76,6 +76,9 @@ class MasterSnapshotCoordinator : public tablet::SnapshotCoordinator {
 
   CHECKED_STATUS FillHeartbeatResponse(TSHeartbeatResponsePB* resp);
 
+  // For each returns map from schedule id to sorted vectors of tablets id in this schedule.
+  Result<SnapshotSchedulesToTabletsMap> MakeSnapshotSchedulesToTabletsMap();
+
   void Start();
 
   void Shutdown();
diff --git a/src/yb/master/snapshot_coordinator_context.h b/src/yb/master/snapshot_coordinator_context.h
index f6c30b920040..d08241e793fe 100644
--- a/src/yb/master/snapshot_coordinator_context.h
+++ b/src/yb/master/snapshot_coordinator_context.h
@@ -63,7 +63,8 @@ class SnapshotCoordinatorContext {
   virtual Result<SysRowEntries> CollectEntries(
       const google::protobuf::RepeatedPtrField<TableIdentifierPB>& tables,
       bool add_indexes,
-      bool include_parent_colocated_table) = 0;
+      bool include_parent_colocated_table,
+      bool succeed_if_create_in_progress) = 0;
 
   virtual const Schema& schema() = 0;
 
diff --git a/src/yb/master/snapshot_schedule_state.cc b/src/yb/master/snapshot_schedule_state.cc
index e6f9927ff0e2..b86f6acd4954 100644
--- a/src/yb/master/snapshot_schedule_state.cc
+++ b/src/yb/master/snapshot_schedule_state.cc
@@ -16,6 +16,7 @@
 #include "yb/docdb/docdb.pb.h"
 #include "yb/docdb/key_bytes.h"
 
+#include "yb/master/catalog_entity_info.h"
 #include "yb/master/snapshot_coordinator_context.h"
 
 #include "yb/util/pb_util.h"
diff --git a/src/yb/master/sys_catalog.cc b/src/yb/master/sys_catalog.cc
index 66cd0a9ece04..d8ef01bf6b3e 100644
--- a/src/yb/master/sys_catalog.cc
+++ b/src/yb/master/sys_catalog.cc
@@ -221,8 +221,7 @@ Status SysCatalogTable::CreateAndFlushConsensusMeta(
 Status SysCatalogTable::Load(FsManager* fs_manager) {
   LOG(INFO) << "Trying to load previous SysCatalogTable data from disk";
   // Load Metadata Information from disk
-  scoped_refptr<tablet::RaftGroupMetadata> metadata;
-  RETURN_NOT_OK(tablet::RaftGroupMetadata::Load(fs_manager, kSysCatalogTabletId, &metadata));
+  auto metadata = VERIFY_RESULT(tablet::RaftGroupMetadata::Load(fs_manager, kSysCatalogTabletId));
 
   // Verify that the schema is the current one
   if (!metadata->schema()->Equals(schema_)) {
@@ -287,7 +286,6 @@ Status SysCatalogTable::Load(FsManager* fs_manager) {
 Status SysCatalogTable::CreateNew(FsManager *fs_manager) {
   LOG(INFO) << "Creating new SysCatalogTable data";
   // Create the new Metadata
-  scoped_refptr<tablet::RaftGroupMetadata> metadata;
   Schema schema = BuildTableSchema();
   PartitionSchema partition_schema;
   RETURN_NOT_OK(PartitionSchema::FromPB(PartitionSchemaPB(), schema, &partition_schema));
@@ -297,21 +295,16 @@ Status SysCatalogTable::CreateNew(FsManager *fs_manager) {
   RETURN_NOT_OK(partition_schema.CreatePartitions(split_rows, schema, &partitions));
   DCHECK_EQ(1, partitions.size());
 
-  RETURN_NOT_OK(tablet::RaftGroupMetadata::CreateNew(
-      fs_manager,
-      kSysCatalogTableId,
-      kSysCatalogTabletId,
-      "",
-      table_name(),
-      TableType::YQL_TABLE_TYPE,
-      schema,
-      IndexMap(),
-      partition_schema,
-      partitions[0],
-      boost::none /* index_info */,
-      0 /* schema_version */,
-      tablet::TABLET_DATA_READY,
-      &metadata));
+  auto table_info = std::make_shared<tablet::TableInfo>(
+      kSysCatalogTableId, "", table_name(), TableType::YQL_TABLE_TYPE, schema, IndexMap(),
+      boost::none /* index_info */, 0 /* schema_version */, partition_schema);
+  auto metadata = VERIFY_RESULT(tablet::RaftGroupMetadata::CreateNew(tablet::RaftGroupMetadataData {
+    .fs_manager = fs_manager,
+    .table_info = table_info,
+    .raft_group_id = kSysCatalogTabletId,
+    .partition = partitions[0],
+    .tablet_data_state = tablet::TABLET_DATA_READY,
+  }));
 
   RaftConfigPB config;
   RETURN_NOT_OK_PREPEND(SetupConfig(master_->opts(), &config),
diff --git a/src/yb/tablet/tablet-harness.h b/src/yb/tablet/tablet-harness.h
index 91fb407f5f61..c74b7f7c4ec1 100644
--- a/src/yb/tablet/tablet-harness.h
+++ b/src/yb/tablet/tablet-harness.h
@@ -109,20 +109,16 @@ class TabletHarness {
     }
     RETURN_NOT_OK(fs_manager_->Open());
 
-    RaftGroupMetadataPtr metadata;
-    RETURN_NOT_OK(RaftGroupMetadata::LoadOrCreate(
-        fs_manager_.get(),
-        "YBTableTest",
-        options_.tablet_id,
-        "test",
-        "YBTableTest",
-        options_.table_type,
-        schema_,
-        partition.first,
-        partition.second,
-        boost::none /* index_info */,
-        TABLET_DATA_READY,
-        &metadata));
+    auto table_info = std::make_shared<TableInfo>(
+        "YBTableTest", "test", "YBTableTest", options_.table_type, schema_, IndexMap(), boost::none,
+        0 /* schema_version */, partition.first);
+    auto metadata = VERIFY_RESULT(RaftGroupMetadata::LoadOrCreate(RaftGroupMetadataData {
+      .fs_manager = fs_manager_.get(),
+      .table_info = table_info,
+      .raft_group_id = options_.tablet_id,
+      .partition = partition.second,
+      .tablet_data_state = TABLET_DATA_READY,
+    }));
     if (options_.enable_metrics) {
       metrics_registry_.reset(new MetricRegistry());
     }
@@ -139,8 +135,7 @@ class TabletHarness {
   }
 
   Result<TabletPtr> OpenTablet(const TabletId& tablet_id) {
-    RaftGroupMetadataPtr metadata;
-    RETURN_NOT_OK(RaftGroupMetadata::Load(fs_manager_.get(), tablet_id, &metadata));
+    auto metadata = VERIFY_RESULT(RaftGroupMetadata::Load(fs_manager_.get(), tablet_id));
     TabletOptions tablet_options;
     auto tablet = std::make_shared<Tablet>(MakeTabletInitData(metadata));
     RETURN_NOT_OK(tablet->Open());
diff --git a/src/yb/tablet/tablet_bootstrap-test.cc b/src/yb/tablet/tablet_bootstrap-test.cc
index ad5bd1c6b5e1..566de78cf61b 100644
--- a/src/yb/tablet/tablet_bootstrap-test.cc
+++ b/src/yb/tablet/tablet_bootstrap-test.cc
@@ -161,11 +161,10 @@ struct BootstrapTestHooksImpl : public TabletBootstrapTestHooksIf {
   bool transactional = false;
 };
 
+static constexpr TableType kTableType = TableType::YQL_TABLE_TYPE;
+
 class BootstrapTest : public LogTestBase {
  protected:
-
-  static constexpr TableType kTableType = TableType::YQL_TABLE_TYPE;
-
   void SetUp() override {
     LogTestBase::SetUp();
     test_hooks_ = std::make_shared<BootstrapTestHooksImpl>();
@@ -175,19 +174,16 @@ class BootstrapTest : public LogTestBase {
     Schema schema = SchemaBuilder(schema_).Build();
     std::pair<PartitionSchema, Partition> partition = CreateDefaultPartition(schema);
 
-    RETURN_NOT_OK(RaftGroupMetadata::LoadOrCreate(
-        fs_manager_.get(),
-        log::kTestTable,
-        log::kTestTablet,
-        log::kTestNamespace,
-        log::kTestTable,
-        kTableType,
-        schema,
-        partition.first,
-        partition.second,
-        boost::none /* index_info */,
-        TABLET_DATA_READY,
-        meta));
+    auto table_info = std::make_shared<TableInfo>(
+        log::kTestTable, log::kTestNamespace, log::kTestTable, kTableType, schema, IndexMap(),
+        boost::none /* index_info */, 0 /* schema_version */, partition.first);
+    *meta = VERIFY_RESULT(RaftGroupMetadata::LoadOrCreate(RaftGroupMetadataData {
+      .fs_manager = fs_manager_.get(),
+      .table_info = table_info,
+      .raft_group_id = log::kTestTablet,
+      .partition = partition.second,
+      .tablet_data_state = TABLET_DATA_READY,
+    }));
     return (*meta)->Flush();
   }
 
diff --git a/src/yb/tablet/tablet_metadata.cc b/src/yb/tablet/tablet_metadata.cc
index aa15c96ad387..a481b0fd3d41 100644
--- a/src/yb/tablet/tablet_metadata.cc
+++ b/src/yb/tablet/tablet_metadata.cc
@@ -250,27 +250,13 @@ std::string MakeTabletDirName(const TabletId& tablet_id) {
 
 // ============================================================================
 
-Status RaftGroupMetadata::CreateNew(FsManager* fs_manager,
-                                 const TableId& table_id,
-                                 const RaftGroupId& raft_group_id,
-                                 const string& namespace_name,
-                                 const string& table_name,
-                                 const TableType table_type,
-                                 const Schema& schema,
-                                 const IndexMap& index_map,
-                                 const PartitionSchema& partition_schema,
-                                 const Partition& partition,
-                                 const boost::optional<IndexInfo>& index_info,
-                                 const uint32_t schema_version,
-                                 const TabletDataState& initial_tablet_data_state,
-                                 RaftGroupMetadataPtr* metadata,
-                                 const string& data_root_dir,
-                                 const string& wal_root_dir,
-                                 const bool colocated) {
-
+Result<RaftGroupMetadataPtr> RaftGroupMetadata::CreateNew(
+    const RaftGroupMetadataData& data, const std::string& data_root_dir,
+    const std::string& wal_root_dir) {
+  auto* fs_manager = data.fs_manager;
   // Verify that no existing Raft group exists with the same ID.
-  if (fs_manager->env()->FileExists(fs_manager->GetRaftGroupMetadataPath(raft_group_id))) {
-    return STATUS(AlreadyPresent, "Raft group already exists", raft_group_id);
+  if (fs_manager->env()->FileExists(fs_manager->GetRaftGroupMetadataPath(data.raft_group_id))) {
+    return STATUS(AlreadyPresent, "Raft group already exists", data.raft_group_id);
   }
 
   auto wal_top_dir = wal_root_dir;
@@ -289,70 +275,40 @@ Status RaftGroupMetadata::CreateNew(FsManager* fs_manager,
     wal_top_dir = wal_root_dirs[rand.Uniform(wal_root_dirs.size())];
   }
 
-  const string table_dir_name = Substitute("table-$0", table_id);
-  const string tablet_dir_name = MakeTabletDirName(raft_group_id);
+  const string table_dir_name = Substitute("table-$0", data.table_info->table_id);
+  const string tablet_dir_name = MakeTabletDirName(data.raft_group_id);
   const string wal_dir = JoinPathSegments(wal_top_dir, table_dir_name, tablet_dir_name);
   const string rocksdb_dir = JoinPathSegments(
       data_top_dir, FsManager::kRocksDBDirName, table_dir_name, tablet_dir_name);
 
-  RaftGroupMetadataPtr ret(new RaftGroupMetadata(fs_manager,
-                                                       table_id,
-                                                       raft_group_id,
-                                                       namespace_name,
-                                                       table_name,
-                                                       table_type,
-                                                       rocksdb_dir,
-                                                       wal_dir,
-                                                       schema,
-                                                       index_map,
-                                                       partition_schema,
-                                                       partition,
-                                                       index_info,
-                                                       schema_version,
-                                                       initial_tablet_data_state,
-                                                       colocated));
+  RaftGroupMetadataPtr ret(new RaftGroupMetadata(data, rocksdb_dir, wal_dir));
   RETURN_NOT_OK(ret->Flush());
-  metadata->swap(ret);
-  return Status::OK();
+  return ret;
 }
 
-Status RaftGroupMetadata::Load(FsManager* fs_manager,
-                            const RaftGroupId& raft_group_id,
-                            RaftGroupMetadataPtr* metadata) {
+Result<RaftGroupMetadataPtr> RaftGroupMetadata::Load(
+    FsManager* fs_manager, const RaftGroupId& raft_group_id) {
   RaftGroupMetadataPtr ret(new RaftGroupMetadata(fs_manager, raft_group_id));
   RETURN_NOT_OK(ret->LoadFromDisk());
-  metadata->swap(ret);
-  return Status::OK();
+  return ret;
 }
 
-Status RaftGroupMetadata::LoadOrCreate(FsManager* fs_manager,
-                                    const string& table_id,
-                                    const RaftGroupId& raft_group_id,
-                                    const string& namespace_name,
-                                    const string& table_name,
-                                    TableType table_type,
-                                    const Schema& schema,
-                                    const PartitionSchema& partition_schema,
-                                    const Partition& partition,
-                                    const boost::optional<IndexInfo>& index_info,
-                                    const TabletDataState& initial_tablet_data_state,
-                                    RaftGroupMetadataPtr* metadata) {
-  Status s = Load(fs_manager, raft_group_id, metadata);
-  if (s.ok()) {
-    if (!(**metadata).schema()->Equals(schema)) {
+Result<RaftGroupMetadataPtr> RaftGroupMetadata::LoadOrCreate(const RaftGroupMetadataData& data) {
+  auto metadata = Load(data.fs_manager, data.raft_group_id);
+  if (metadata.ok()) {
+    if (!(**metadata).schema()->Equals(data.table_info->schema)) {
       return STATUS(Corruption, Substitute("Schema on disk ($0) does not "
         "match expected schema ($1)", (*metadata)->schema()->ToString(),
-        schema.ToString()));
+        data.table_info->schema.ToString()));
     }
-    return Status::OK();
-  } else if (s.IsNotFound()) {
-    return CreateNew(
-        fs_manager, table_id, raft_group_id, namespace_name, table_name, table_type, schema,
-        IndexMap(), partition_schema, partition, index_info, 0 /* schema_version */,
-        initial_tablet_data_state, metadata);
-  } else {
-    return s;
+    return *metadata;
   }
+
+  if (metadata.status().IsNotFound()) {
+    return CreateNew(data);
+  }
+
+  return metadata.status();
 }
 
 template <class TablesMap>
@@ -497,46 +453,21 @@ Status RaftGroupMetadata::DeleteSuperBlock() {
   return Status::OK();
 }
 
-RaftGroupMetadata::RaftGroupMetadata(FsManager* fs_manager,
-                               TableId table_id,
-                               RaftGroupId raft_group_id,
-                               string namespace_name,
-                               string table_name,
-                               TableType table_type,
-                               const string rocksdb_dir,
-                               const string wal_dir,
-                               const Schema& schema,
-                               const IndexMap& index_map,
-                               PartitionSchema partition_schema,
-                               Partition partition,
-                               const boost::optional<IndexInfo>& index_info,
-                               const uint32_t schema_version,
-                               const TabletDataState& tablet_data_state,
-                               const bool colocated)
+RaftGroupMetadata::RaftGroupMetadata(
+    const RaftGroupMetadataData& data, const std::string& data_dir, const std::string& wal_dir)
     : state_(kNotWrittenYet),
-      raft_group_id_(std::move(raft_group_id)),
-      partition_(std::make_shared<Partition>(std::move(partition))),
-      primary_table_id_(table_id),
-      kv_store_(KvStoreId(raft_group_id_), rocksdb_dir),
-      fs_manager_(fs_manager),
+      raft_group_id_(data.raft_group_id),
+      partition_(std::make_shared<Partition>(data.partition)),
+      primary_table_id_(data.table_info->table_id),
+      kv_store_(KvStoreId(raft_group_id_), data_dir, data.snapshot_schedules),
+      fs_manager_(data.fs_manager),
       wal_dir_(wal_dir),
-      tablet_data_state_(tablet_data_state),
-      colocated_(colocated),
+      tablet_data_state_(data.tablet_data_state),
+      colocated_(data.colocated),
       cdc_min_replicated_index_(std::numeric_limits<int64_t>::max()) {
-  CHECK(schema.has_column_ids());
-  CHECK_GT(schema.num_key_columns(), 0);
-  kv_store_.tables.emplace(
-      primary_table_id_,
-      std::make_shared<TableInfo>(
-          std::move(table_id),
-          std::move(namespace_name),
-          std::move(table_name),
-          table_type,
-          schema,
-          index_map,
-          index_info,
-          schema_version,
-          std::move(partition_schema)));
+  CHECK(data.table_info->schema.has_column_ids());
+  CHECK_GT(data.table_info->schema.num_key_columns(), 0);
+  kv_store_.tables.emplace(primary_table_id_, data.table_info);
 }
 
 RaftGroupMetadata::~RaftGroupMetadata() {
diff --git a/src/yb/tablet/tablet_metadata.h b/src/yb/tablet/tablet_metadata.h
index ca88b009c042..fe5574489b4b 100644
--- a/src/yb/tablet/tablet_metadata.h
+++ b/src/yb/tablet/tablet_metadata.h
@@ -127,9 +127,11 @@ struct TableInfo {
 struct KvStoreInfo {
   explicit KvStoreInfo(const KvStoreId& kv_store_id_) : kv_store_id(kv_store_id_) {}
 
-  KvStoreInfo(const KvStoreId& kv_store_id_, const std::string& rocksdb_dir_)
+  KvStoreInfo(const KvStoreId& kv_store_id_, const std::string& rocksdb_dir_,
+              const std::vector<SnapshotScheduleId>& snapshot_schedules_)
       : kv_store_id(kv_store_id_),
-        rocksdb_dir(rocksdb_dir_) {}
+        rocksdb_dir(rocksdb_dir_),
+        snapshot_schedules(snapshot_schedules_.begin(), snapshot_schedules_.end()) {}
 
   CHECKED_STATUS LoadFromPB(const KvStoreInfoPB& pb, const TableId& primary_table_id);
 
@@ -161,6 +163,16 @@ struct KvStoreInfo {
   std::unordered_set<SnapshotScheduleId, SnapshotScheduleIdHash> snapshot_schedules;
 };
 
+struct RaftGroupMetadataData {
+  FsManager* fs_manager;
+  TableInfoPtr table_info;
+  RaftGroupId raft_group_id;
+  Partition partition;
+  TabletDataState tablet_data_state;
+  bool colocated = false;
+  std::vector<SnapshotScheduleId> snapshot_schedules;
+};
+
 // At startup, the TSTabletManager will load a RaftGroupMetadata for each
 // super block found in the tablets/ directory, and then instantiate
 // Raft groups from this data.
@@ -172,46 +184,19 @@ class RaftGroupMetadata : public RefCountedThreadSafe<RaftGroupMetadata> {
   // data_root_dir and wal_root_dir dictates which disk this Raft group will
   // use in the respective directories.
   // If empty string is passed in, it will be randomly chosen.
-  static CHECKED_STATUS CreateNew(FsManager* fs_manager,
-                                  const std::string& table_id,
-                                  const RaftGroupId& raft_group_id,
-                                  const std::string& namespace_name,
-                                  const std::string& table_name,
-                                  const TableType table_type,
-                                  const Schema& schema,
-                                  const IndexMap& index_map,
-                                  const PartitionSchema& partition_schema,
-                                  const Partition& partition,
-                                  const boost::optional<IndexInfo>& index_info,
-                                  const uint32_t schema_version,
-                                  const TabletDataState& initial_tablet_data_state,
-                                  RaftGroupMetadataPtr* metadata,
-                                  const std::string& data_root_dir = std::string(),
-                                  const std::string& wal_root_dir = std::string(),
-                                  const bool colocated = false);
+  static Result<RaftGroupMetadataPtr> CreateNew(
+      const RaftGroupMetadataData& data, const std::string& data_root_dir = std::string(),
+      const std::string& wal_root_dir = std::string());
 
   // Load existing metadata from disk.
-  static CHECKED_STATUS Load(FsManager* fs_manager,
-                             const RaftGroupId& raft_group_id,
-                             RaftGroupMetadataPtr* metadata);
+  static Result<RaftGroupMetadataPtr> Load(FsManager* fs_manager, const RaftGroupId& raft_group_id);
 
   // Try to load an existing Raft group. If it does not exist, create it.
   // If it already existed, verifies that the schema of the Raft group matches the
   // provided 'schema'.
   //
   // This is mostly useful for tests which instantiate Raft groups directly.
-  static CHECKED_STATUS LoadOrCreate(FsManager* fs_manager,
-                                     const std::string& table_id,
-                                     const RaftGroupId& raft_group_id,
-                                     const std::string& namespace_name,
-                                     const std::string& table_name,
-                                     const TableType table_type,
-                                     const Schema& schema,
-                                     const PartitionSchema& partition_schema,
-                                     const Partition& partition,
-                                     const boost::optional<IndexInfo>& index_info,
-                                     const TabletDataState& initial_tablet_data_state,
-                                     RaftGroupMetadataPtr* metadata);
+  static Result<RaftGroupMetadataPtr> LoadOrCreate(const RaftGroupMetadataData& data);
 
   Result<TableInfoPtr> GetTableInfo(const TableId& table_id) const;
   Result<TableInfoPtr> GetTableInfoUnlocked(const TableId& table_id) const;
@@ -499,25 +484,9 @@ class RaftGroupMetadata : public RefCountedThreadSafe<RaftGroupMetadata> {
   ~RaftGroupMetadata();
 
   // Constructor for creating a new Raft group.
-  //
-  // TODO: get rid of this many-arg constructor in favor of just passing in a
-  // SuperBlock, which already contains all of these fields.
-  RaftGroupMetadata(FsManager* fs_manager,
-                    TableId table_id,
-                    RaftGroupId raft_group_id,
-                    std::string namespace_name,
-                    std::string table_name,
-                    TableType table_type,
-                    const std::string rocksdb_dir,
-                    const std::string wal_dir,
-                    const Schema& schema,
-                    const IndexMap& index_map,
-                    PartitionSchema partition_schema,
-                    Partition partition,
-                    const boost::optional<IndexInfo>& index_info,
-                    const uint32_t schema_version,
-                    const TabletDataState& tablet_data_state,
-                    const bool colocated = false);
+  explicit RaftGroupMetadata(
+      const RaftGroupMetadataData& data, const std::string& data_dir,
+      const std::string& wal_dir);
 
   // Constructor for loading an existing Raft group.
   RaftGroupMetadata(FsManager* fs_manager, RaftGroupId raft_group_id);
diff --git a/src/yb/tools/fs_tool.cc b/src/yb/tools/fs_tool.cc
index 7cdd5832a51c..3c09a8122221 100644
--- a/src/yb/tools/fs_tool.cc
+++ b/src/yb/tools/fs_tool.cc
@@ -157,8 +157,7 @@ Status FsTool::ListAllLogSegments() {
 Status FsTool::ListLogSegmentsForTablet(const string& tablet_id) {
   DCHECK(initialized_);
 
-  RaftGroupMetadataPtr meta;
-  RETURN_NOT_OK(RaftGroupMetadata::Load(fs_manager_.get(), tablet_id, &meta));
+  auto meta = VERIFY_RESULT(RaftGroupMetadata::Load(fs_manager_.get(), tablet_id));
 
   const string& tablet_wal_dir = meta->wal_dir();
   if (!fs_manager_->Exists(tablet_wal_dir)) {
@@ -238,8 +237,7 @@ Status FsTool::PrintLogSegmentHeader(const string& path,
 }
 
 Status FsTool::PrintTabletMeta(const string& tablet_id, int indent) {
-  RaftGroupMetadataPtr meta;
-  RETURN_NOT_OK(RaftGroupMetadata::Load(fs_manager_.get(), tablet_id, &meta));
+  auto meta = VERIFY_RESULT(RaftGroupMetadata::Load(fs_manager_.get(), tablet_id));
 
   const SchemaPtr schema = meta->schema();
 
@@ -261,8 +259,7 @@ Status FsTool::PrintTabletMeta(const string& tablet_id, int indent) {
 Status FsTool::DumpTabletData(const std::string& tablet_id) {
   DCHECK(initialized_);
 
-  RaftGroupMetadataPtr meta;
-  RETURN_NOT_OK(RaftGroupMetadata::Load(fs_manager_.get(), tablet_id, &meta));
+  auto meta = VERIFY_RESULT(RaftGroupMetadata::Load(fs_manager_.get(), tablet_id));
 
   scoped_refptr<log::LogAnchorRegistry> reg(new log::LogAnchorRegistry());
   tablet::TabletOptions tablet_options;
diff --git a/src/yb/tserver/mini_tablet_server.cc b/src/yb/tserver/mini_tablet_server.cc
index 7283a288a15f..9ce17488ba81 100644
--- a/src/yb/tserver/mini_tablet_server.cc
+++ b/src/yb/tserver/mini_tablet_server.cc
@@ -286,9 +286,12 @@ Status MiniTabletServer::AddTestTablet(const std::string& ns_id,
   Schema schema_with_ids = SchemaBuilder(schema).Build();
   pair<PartitionSchema, Partition> partition = tablet::CreateDefaultPartition(schema_with_ids);
 
-  return server_->tablet_manager()->CreateNewTablet(
-      table_id, tablet_id, partition.second, ns_id, table_id, table_type, schema_with_ids,
-      partition.first, boost::none /* index_info */, config, nullptr);
+  auto table_info = std::make_shared<tablet::TableInfo>(
+      table_id, ns_id, table_id, table_type, schema_with_ids, IndexMap(),
+      boost::none /* index_info */, 0 /* schema_version */, partition.first);
+
+  return ResultToStatus(server_->tablet_manager()->CreateNewTablet(
+      table_info, tablet_id, partition.second, config));
 }
 
 void MiniTabletServer::FailHeartbeats(bool fail_heartbeats_for_tests) {
diff --git a/src/yb/tserver/remote_bootstrap_client.cc b/src/yb/tserver/remote_bootstrap_client.cc
index f2de30af2611..4ed452e2fa1f 100644
--- a/src/yb/tserver/remote_bootstrap_client.cc
+++ b/src/yb/tserver/remote_bootstrap_client.cc
@@ -330,31 +330,23 @@ Status RemoteBootstrapClient::Start(const string& bootstrap_peer_uuid,
                                               &data_root_dir,
                                               &wal_root_dir);
     }
-    Status create_status = RaftGroupMetadata::CreateNew(
-        &fs_manager(),
-        table_id,
-        tablet_id_,
-        table.namespace_name(),
-        table.table_name(),
-        table.table_type(),
-        schema,
+    auto table_info = std::make_shared<tablet::TableInfo>(
+        table_id, table.namespace_name(), table.table_name(), table.table_type(), schema,
         IndexMap(table.indexes()),
-        partition_schema,
-        partition,
         table.has_index_info() ? boost::optional<IndexInfo>(table.index_info()) : boost::none,
-        table.schema_version(),
-        tablet::TABLET_DATA_COPYING,
-        &meta_,
-        data_root_dir,
-        wal_root_dir,
-        colocated);
-    if (ts_manager != nullptr && !create_status.ok()) {
-      ts_manager->UnregisterDataWalDir(table_id,
-                                       tablet_id_,
-                                       data_root_dir,
-                                       wal_root_dir);
+        table.schema_version(), partition_schema);
+    auto create_result = RaftGroupMetadata::CreateNew(tablet::RaftGroupMetadataData {
+        .fs_manager = &fs_manager(),
+        .table_info = table_info,
+        .raft_group_id = tablet_id_,
+        .partition = partition,
+        .tablet_data_state = tablet::TABLET_DATA_COPYING,
+        .colocated = colocated }, data_root_dir, wal_root_dir);
+    if (ts_manager != nullptr && !create_result.ok()) {
+      ts_manager->UnregisterDataWalDir(table_id, tablet_id_, data_root_dir, wal_root_dir);
     }
-    RETURN_NOT_OK(create_status);
+    RETURN_NOT_OK(create_result);
+    meta_ = std::move(*create_result);
 
     vector<DeletedColumn> deleted_cols;
     for (const DeletedColumnPB& col_pb : table.deleted_cols()) {
diff --git a/src/yb/tserver/tablet_server-test.cc b/src/yb/tserver/tablet_server-test.cc
index 7bea632d1011..cfca94aa125b 100644
--- a/src/yb/tserver/tablet_server-test.cc
+++ b/src/yb/tserver/tablet_server-test.cc
@@ -804,9 +804,11 @@ TEST_F(TabletServerTest, TestWriteOutOfBounds) {
   CHECK_OK(PartitionSchema::FromPB(PartitionSchemaPB(), schema, &partition_schema));
 
   Partition partition;
+  auto table_info = std::make_shared<tablet::TableInfo>(
+      "TestWriteOutOfBoundsTable", "test_ns", tabletId, YQL_TABLE_TYPE, schema, IndexMap(),
+      boost::none /* index_info */, 0 /* schema_version */, partition_schema);
   ASSERT_OK(mini_server_->server()->tablet_manager()->CreateNewTablet(
-      "TestWriteOutOfBoundsTable", tabletId, partition, "test_ns", tabletId, YQL_TABLE_TYPE, schema,
-      partition_schema, boost::none /* index_info */, mini_server_->CreateLocalConfig(), nullptr));
+      table_info, tabletId, partition, mini_server_->CreateLocalConfig()));
 
   ASSERT_OK(WaitForTabletRunning(tabletId));
 
diff --git a/src/yb/tserver/tablet_service.cc b/src/yb/tserver/tablet_service.cc
index 813fab61e217..a3ffe1c1cfdd 100644
--- a/src/yb/tserver/tablet_service.cc
+++ b/src/yb/tserver/tablet_service.cc
@@ -1099,35 +1099,41 @@ void TabletServiceImpl::Truncate(const TruncateRequestPB* req,
 void TabletServiceAdminImpl::CreateTablet(const CreateTabletRequestPB* req,
                                           CreateTabletResponsePB* resp,
                                           rpc::RpcContext context) {
+  if (!CheckUuidMatchOrRespond(server_->tablet_manager(), "CreateTablet", req, resp, &context)) {
+    return;
+  }
+  auto status = DoCreateTablet(req, resp);
+  if (!status.ok()) {
+    auto code = TabletServerError(status).value();
+    if (!code) {
+      code = TabletServerErrorPB::UNKNOWN_ERROR;
+    }
+    SetupErrorAndRespond(resp->mutable_error(), status, code, &context);
+  } else {
+    context.RespondSuccess();
+  }
+}
+
+Status TabletServiceAdminImpl::DoCreateTablet(const CreateTabletRequestPB* req,
+                                              CreateTabletResponsePB* resp) {
   if (PREDICT_FALSE(FLAGS_TEST_txn_status_table_tablet_creation_delay_ms > 0 &&
                     req->table_type() == TableType::TRANSACTION_STATUS_TABLE_TYPE)) {
     std::this_thread::sleep_for(FLAGS_TEST_txn_status_table_tablet_creation_delay_ms * 1ms);
   }
 
-  if (!CheckUuidMatchOrRespond(server_->tablet_manager(), "CreateTablet", req, resp, &context)) {
-    return;
-  }
   DVLOG(3) << "Received CreateTablet RPC: " << yb::ToString(*req);
   TRACE_EVENT1("tserver", "CreateTablet",
                "tablet_id", req->tablet_id());
 
   Schema schema;
-  Status s = SchemaFromPB(req->schema(), &schema);
-  DCHECK(schema.has_column_ids());
-  if (!s.ok()) {
-    SetupErrorAndRespond(resp->mutable_error(),
-                         STATUS(InvalidArgument, "Invalid Schema."),
-                         TabletServerErrorPB::INVALID_SCHEMA, &context);
-    return;
-  }
-
   PartitionSchema partition_schema;
-  s = PartitionSchema::FromPB(req->partition_schema(), schema, &partition_schema);
-  if (!s.ok()) {
-    SetupErrorAndRespond(resp->mutable_error(),
-                         STATUS(InvalidArgument, "Invalid PartitionSchema."),
-                         TabletServerErrorPB::INVALID_SCHEMA, &context);
-    return;
+  auto status = SchemaFromPB(req->schema(), &schema);
+  if (status.ok()) {
+    DCHECK(schema.has_column_ids());
+    status = PartitionSchema::FromPB(req->partition_schema(), schema, &partition_schema);
+  }
+  if (!status.ok()) {
+    return status.CloneAndAddErrorCode(TabletServerError(TabletServerErrorPB::INVALID_SCHEMA));
   }
 
   Partition partition;
@@ -1139,22 +1145,25 @@ void TabletServiceAdminImpl::CreateTablet(const CreateTabletRequestPB* req,
             << partition_schema.PartitionDebugString(partition, schema);
   VLOG(1) << "Full request: " << req->DebugString();
 
-  s = server_->tablet_manager()->CreateNewTablet(
-      req->table_id(), req->tablet_id(), partition, req->namespace_name(), req->table_name(),
-      req->table_type(), schema, partition_schema,
+  auto table_info = std::make_shared<tablet::TableInfo>(
+      req->table_id(), req->namespace_name(), req->table_name(), req->table_type(), schema,
+      IndexMap(),
       req->has_index_info() ? boost::optional<IndexInfo>(req->index_info()) : boost::none,
-      req->config(), /* tablet_peer */ nullptr, req->colocated());
-  if (PREDICT_FALSE(!s.ok())) {
-    TabletServerErrorPB::Code code;
-    if (s.IsAlreadyPresent()) {
-      code = TabletServerErrorPB::TABLET_ALREADY_EXISTS;
-    } else {
-      code = TabletServerErrorPB::UNKNOWN_ERROR;
-    }
-    SetupErrorAndRespond(resp->mutable_error(), s, code, &context);
-    return;
+      0 /* schema_version */, partition_schema);
+  std::vector<SnapshotScheduleId> snapshot_schedules;
+  snapshot_schedules.reserve(req->snapshot_schedules().size());
+  for (const auto& id : req->snapshot_schedules()) {
+    snapshot_schedules.push_back(VERIFY_RESULT(FullyDecodeSnapshotScheduleId(id)));
+  }
+  status = ResultToStatus(server_->tablet_manager()->CreateNewTablet(
+      table_info, req->tablet_id(), partition, req->config(), req->colocated(),
+      snapshot_schedules));
+  if (PREDICT_FALSE(!status.ok())) {
+    return status.IsAlreadyPresent()
+        ? status.CloneAndAddErrorCode(TabletServerError(TabletServerErrorPB::TABLET_ALREADY_EXISTS))
+        : status;
   }
-  context.RespondSuccess();
+  return Status::OK();
 }
 
 void TabletServiceAdminImpl::DeleteTablet(const DeleteTabletRequestPB* req,
diff --git a/src/yb/tserver/tablet_service.h b/src/yb/tserver/tablet_service.h
index a83e01dfd3e4..21e01bf90770 100644
--- a/src/yb/tserver/tablet_service.h
+++ b/src/yb/tserver/tablet_service.h
@@ -248,6 +248,8 @@ class TabletServiceAdminImpl : public TabletServerAdminServiceIf {
  private:
   TabletServer* server_;
 
+  CHECKED_STATUS DoCreateTablet(const CreateTabletRequestPB* req, CreateTabletResponsePB* resp);
+
   // Used to implement wait/signal mechanism for backfill requests.
   // Since the number of concurrently allowed backfill requests is
   // limited.
diff --git a/src/yb/tserver/ts_tablet_manager-test.cc b/src/yb/tserver/ts_tablet_manager-test.cc
index 03c708cf1e61..3a1114371fe7 100644
--- a/src/yb/tserver/ts_tablet_manager-test.cc
+++ b/src/yb/tserver/ts_tablet_manager-test.cc
@@ -123,10 +123,11 @@ class TsTabletManagerTest : public YBTest {
     Schema full_schema = SchemaBuilder(schema).Build();
     std::pair<PartitionSchema, Partition> partition = tablet::CreateDefaultPartition(full_schema);
 
-    std::shared_ptr<tablet::TabletPeer> tablet_peer;
-    RETURN_NOT_OK(tablet_manager_->CreateNewTablet(
-        table_id, tablet_id, partition.second, tablet_id, tablet_id, TableType::DEFAULT_TABLE_TYPE,
-        full_schema, partition.first, boost::none /* index_info */, config_, &tablet_peer));
+    auto table_info = std::make_shared<tablet::TableInfo>(
+        table_id, tablet_id, tablet_id, TableType::DEFAULT_TABLE_TYPE, full_schema, IndexMap(),
+        boost::none /* index_info */, 0 /* schema_version */, partition.first);
+    auto tablet_peer = VERIFY_RESULT(tablet_manager_->CreateNewTablet(
+        table_info, tablet_id, partition.second, config_));
     if (out_tablet_peer) {
       (*out_tablet_peer) = tablet_peer;
     }
diff --git a/src/yb/tserver/ts_tablet_manager.cc b/src/yb/tserver/ts_tablet_manager.cc
index dc71a37cef9a..d2f5f2b5fce9 100644
--- a/src/yb/tserver/ts_tablet_manager.cc
+++ b/src/yb/tserver/ts_tablet_manager.cc
@@ -747,19 +747,13 @@ TSTabletManager::StartTabletStateTransitionForCreation(const TabletId& tablet_id
   return deleter;
 }
 
-Status TSTabletManager::CreateNewTablet(
-    const string& table_id,
+Result<TabletPeerPtr> TSTabletManager::CreateNewTablet(
+    const tablet::TableInfoPtr& table_info,
     const string& tablet_id,
     const Partition& partition,
-    const string& namespace_name,
-    const string& table_name,
-    TableType table_type,
-    const Schema& schema,
-    const PartitionSchema& partition_schema,
-    const boost::optional<IndexInfo>& index_info,
     RaftConfigPB config,
-    TabletPeerPtr* tablet_peer,
-    const bool colocated) {
+    const bool colocated,
+    const std::vector<SnapshotScheduleId>& snapshot_schedules) {
   if (state() != MANAGER_RUNNING) {
     return STATUS_FORMAT(IllegalState, "Manager is not running: $0", state());
   }
@@ -778,33 +772,26 @@ Status TSTabletManager::CreateNewTablet(
 
   // Create the metadata.
   TRACE("Creating new metadata...");
-  RaftGroupMetadataPtr meta;
   string data_root_dir;
   string wal_root_dir;
-  GetAndRegisterDataAndWalDir(fs_manager_, table_id, tablet_id, &data_root_dir, &wal_root_dir);
-  Status create_status = RaftGroupMetadata::CreateNew(fs_manager_,
-                                                   table_id,
-                                                   tablet_id,
-                                                   namespace_name,
-                                                   table_name,
-                                                   table_type,
-                                                   schema,
-                                                   IndexMap(),
-                                                   partition_schema,
-                                                   partition,
-                                                   index_info,
-                                                   0 /* schema_version */,
-                                                   TABLET_DATA_READY,
-                                                   &meta,
-                                                   data_root_dir,
-                                                   wal_root_dir,
-                                                   colocated);
-  if (!create_status.ok()) {
-    UnregisterDataWalDir(table_id, tablet_id, data_root_dir, wal_root_dir);
-  }
-  RETURN_NOT_OK_PREPEND(create_status, "Couldn't create tablet metadata")
+  GetAndRegisterDataAndWalDir(
+      fs_manager_, table_info->table_id, tablet_id, &data_root_dir, &wal_root_dir);
+  auto create_result = RaftGroupMetadata::CreateNew(tablet::RaftGroupMetadataData {
+    .fs_manager = fs_manager_,
+    .table_info = table_info,
+    .raft_group_id = tablet_id,
+    .partition = partition,
+    .tablet_data_state = TABLET_DATA_READY,
+    .colocated = colocated,
+    .snapshot_schedules = snapshot_schedules,
+  }, data_root_dir, wal_root_dir);
+  if (!create_result.ok()) {
+    UnregisterDataWalDir(table_info->table_id, tablet_id, data_root_dir, wal_root_dir);
+  }
+  RETURN_NOT_OK_PREPEND(create_result, "Couldn't create tablet metadata")
+  RaftGroupMetadataPtr meta = std::move(*create_result);
   LOG(INFO) << TabletLogPrefix(tablet_id)
-            << "Created tablet metadata for table: " << table_id;
+            << "Created tablet metadata for table: " << table_info->table_id;
 
   // We must persist the consensus metadata to disk before starting a new
   // tablet's TabletPeer and Consensus implementation.
@@ -818,10 +805,7 @@ Status TSTabletManager::CreateNewTablet(
   RETURN_NOT_OK(
       open_tablet_pool_->SubmitFunc(std::bind(&TSTabletManager::OpenTablet, this, meta, deleter)));
 
-  if (tablet_peer) {
-    *tablet_peer = new_peer;
-  }
-  return Status::OK();
+  return new_peer;
 }
 
 struct TabletCreationMetaData {
@@ -891,8 +875,8 @@ Status TSTabletManager::StartSubtabletsSplit(
     }
 
     // Try to load metadata from previous not completed split.
-    if (RaftGroupMetadata::Load(fs_manager_, subtablet_id, &iter->raft_group_metadata).ok() &&
-        CanServeTabletData(iter->raft_group_metadata->tablet_data_state())) {
+    auto load_result = RaftGroupMetadata::Load(fs_manager_, subtablet_id);
+    if (load_result.ok() && CanServeTabletData(iter->raft_group_metadata->tablet_data_state())) {
       // Sub tablet has been already created and ready during previous split attempt at this node or
       // as a result of remote bootstrap from another node, no need to re-create.
       iter = tcmetas->erase(iter);
@@ -1426,12 +1410,11 @@ Status TSTabletManager::OpenTabletMeta(const string& tablet_id,
                                        RaftGroupMetadataPtr* metadata) {
   LOG(INFO) << "Loading metadata for tablet " << tablet_id;
   TRACE("Loading metadata...");
-  RaftGroupMetadataPtr meta;
-  RETURN_NOT_OK_PREPEND(RaftGroupMetadata::Load(fs_manager_, tablet_id, &meta),
-                        strings::Substitute("Failed to load tablet metadata for tablet id $0",
-                                            tablet_id));
+  auto load_result = RaftGroupMetadata::Load(fs_manager_, tablet_id);
+  RETURN_NOT_OK_PREPEND(load_result,
+                        Format("Failed to load tablet metadata for tablet id $0", tablet_id));
   TRACE("Metadata loaded");
-  metadata->swap(meta);
+  metadata->swap(*load_result);
   return Status::OK();
 }
 
diff --git a/src/yb/tserver/ts_tablet_manager.h b/src/yb/tserver/ts_tablet_manager.h
index 7ffa700eed5b..cf684c69d5be 100644
--- a/src/yb/tserver/ts_tablet_manager.h
+++ b/src/yb/tserver/ts_tablet_manager.h
@@ -165,19 +165,13 @@ class TSTabletManager : public tserver::TabletPeerLookupIf, public tablet::Table
   //
   // If another tablet already exists with this ID, logs a DFATAL
   // and returns a bad Status.
-  CHECKED_STATUS CreateNewTablet(
-      const string& table_id,
+  Result<tablet::TabletPeerPtr> CreateNewTablet(
+      const tablet::TableInfoPtr& table_info,
       const string& tablet_id,
       const Partition& partition,
-      const string& namespace_name,
-      const string& table_name,
-      TableType table_type,
-      const Schema& schema,
-      const PartitionSchema& partition_schema,
-      const boost::optional<IndexInfo>& index_info,
       consensus::RaftConfigPB config,
-      std::shared_ptr<tablet::TabletPeer>* tablet_peer,
-      const bool colocated = false);
+      const bool colocated = false,
+      const std::vector<SnapshotScheduleId>& snapshot_schedules = {});
 
   CHECKED_STATUS ApplyTabletSplit(tablet::SplitOperationState* state, log::Log* raft_log) override;
 
diff --git a/src/yb/tserver/tserver_admin.proto b/src/yb/tserver/tserver_admin.proto
index 1bab792835f1..835005d5b586 100644
--- a/src/yb/tserver/tserver_admin.proto
+++ b/src/yb/tserver/tserver_admin.proto
@@ -184,6 +184,8 @@ message CreateTabletRequestPB {
 
   // True if the tablet is colocated.
   optional bool colocated = 13 [ default = false ];
+
+  repeated bytes snapshot_schedules = 16;
 }
 
 message CreateTabletResponsePB {