diff --git a/Firestore/Example/Firestore.xcodeproj/project.pbxproj b/Firestore/Example/Firestore.xcodeproj/project.pbxproj
index 518d8c3eb68..97e4482ecee 100644
--- a/Firestore/Example/Firestore.xcodeproj/project.pbxproj
+++ b/Firestore/Example/Firestore.xcodeproj/project.pbxproj
@@ -57,8 +57,8 @@
 		08E3D48B3651E4908D75B23A /* async_testing.cc in Sources */ = {isa = PBXBuildFile; fileRef = 872C92ABD71B12784A1C5520 /* async_testing.cc */; };
 		08F44F7DF9A3EF0D35C8FB57 /* FIRNumericTransformTests.mm in Sources */ = {isa = PBXBuildFile; fileRef = D5B25E7E7D6873CBA4571841 /* FIRNumericTransformTests.mm */; };
 		08FA4102AD14452E9587A1F2 /* leveldb_util_test.cc in Sources */ = {isa = PBXBuildFile; fileRef = 332485C4DCC6BA0DBB5E31B7 /* leveldb_util_test.cc */; };
-		0929C73B3F3BFC331E9E9D2F /* resource.pb.cc in Sources */ = {isa = PBXBuildFile; fileRef = 1C3F7302BF4AE6CBC00ECDD0 /* resource.pb.cc */; };
 		095A878BB33211AB52BFAD9F /* leveldb_document_overlay_cache_test.cc in Sources */ = {isa = PBXBuildFile; fileRef = AE89CFF09C6804573841397F /* leveldb_document_overlay_cache_test.cc */; };
+		0929C73B3F3BFC331E9E9D2F /* resource.pb.cc in Sources */ = {isa = PBXBuildFile; fileRef = 1C3F7302BF4AE6CBC00ECDD0 /* resource.pb.cc */; };
 		0963F6D7B0F9AE1E24B82866 /* path_test.cc in Sources */ = {isa = PBXBuildFile; fileRef = 403DBF6EFB541DFD01582AA3 /* path_test.cc */; };
 		096BA3A3703AC1491F281618 /* index.pb.cc in Sources */ = {isa = PBXBuildFile; fileRef = 395E8B07639E69290A929695 /* index.pb.cc */; };
 		098191405BA24F9A7E4F80C6 /* append_only_list_test.cc in Sources */ = {isa = PBXBuildFile; fileRef = 5477CDE922EE71C8000FCC1E /* append_only_list_test.cc */; };
diff --git a/Firestore/core/src/local/index_manager.h b/Firestore/core/src/local/index_manager.h
index c0fe9401b60..bfd98a0c134 100644
--- a/Firestore/core/src/local/index_manager.h
+++ b/Firestore/core/src/local/index_manager.h
@@ -20,10 +20,20 @@
 #include <string>
 #include <vector>
 
+#include "Firestore/core/src/model/model_fwd.h"
+
 namespace firebase {
 namespace firestore {
 
+namespace core {
+class Target;
+class SortedMap;
+}  // namespace core
+
 namespace model {
+class DocumentKey;
+class FieldIndex;
+class IndexOffset;
 class ResourcePath;
 }  // namespace model
 
@@ -39,6 +49,9 @@ class IndexManager {
  public:
   virtual ~IndexManager() = default;
 
+  /** Initializes the IndexManager. */
+  virtual void Start() = 0;
+
   /**
    * Creates an index entry mapping the collection_id (last segment of the path)
    * to the parent path (either the containing document location or the empty
@@ -58,6 +71,65 @@ class IndexManager {
    */
   virtual std::vector<model::ResourcePath> GetCollectionParents(
       const std::string& collection_id) = 0;
+
+  /**
+   * Adds a field path index.
+   *
+   * The actual entries for this index will be created and persisted in the
+   * background by the SDK, and the index will be used for query execution once
+   * values are persisted.
+   */
+  virtual void AddFieldIndex(const model::FieldIndex& index) = 0;
+
+  /** Removes the given field index and deletes all index values. */
+  virtual void DeleteFieldIndex(const model::FieldIndex& index) = 0;
+
+  /**
+   * Returns a list of field indexes that correspond to the specified collection
+   * group.
+   */
+  virtual std::vector<model::FieldIndex> GetFieldIndexes(
+      const std::string& collection_group) = 0;
+
+  /** Returns all configured field indexes. */
+  virtual std::vector<model::FieldIndex> GetFieldIndexes() = 0;
+
+  /**
+   * Returns an index that can be used to serve the provided target. Returns
+   * `nullopt` if no index is configured.
+   */
+  virtual absl::optional<model::FieldIndex> GetFieldIndex(
+      core::Target target) = 0;
+
+  /**
+   * Returns the documents that match the given target based on the provided
+   * index, or `nullopt` if the query cannot be served from an index.
+   */
+  virtual absl::optional<std::vector<model::DocumentKey>>
+  GetDocumentsMatchingTarget(model::FieldIndex fieldIndex,
+                             core::Target target) = 0;
+
+  /**
+   * Returns the next collection group to update. Returns `nullopt` if no
+   * group exists.
+   */
+  virtual absl::optional<std::string> GetNextCollectionGroupToUpdate() = 0;
+
+  /**
+   * Sets the collection group's latest read time.
+   *
+   * This method updates the index offset for all field indices for the
+   * collection group and increments their sequence number.
+   *
+   * Subsequent calls to `GetNextCollectionGroupToUpdate()` will return a
+   * different collection group (unless only one collection group is
+   * configured).
+   */
+  virtual void UpdateCollectionGroup(const std::string& collection_group,
+                                     model::IndexOffset offset) = 0;
+
+  /** Updates the index entries for the provided documents. */
+  virtual void UpdateIndexEntries(const model::DocumentMap& documents) = 0;
 };
 
 }  // namespace local
diff --git a/Firestore/core/src/local/leveldb_index_manager.cc b/Firestore/core/src/local/leveldb_index_manager.cc
index 9cbf11eb1e8..3b4bfd1dd69 100644
--- a/Firestore/core/src/local/leveldb_index_manager.cc
+++ b/Firestore/core/src/local/leveldb_index_manager.cc
@@ -16,23 +16,77 @@
 
 #include "Firestore/core/src/local/leveldb_index_manager.h"
 
+#include <algorithm>
 #include <string>
+#include <utility>
 #include <vector>
 
 #include "Firestore/core/src/local/leveldb_key.h"
 #include "Firestore/core/src/local/leveldb_persistence.h"
-#include "Firestore/core/src/local/memory_index_manager.h"
+#include "Firestore/core/src/local/local_serializer.h"
+#include "Firestore/core/src/model/field_index.h"
+#include "Firestore/core/src/model/model_fwd.h"
 #include "Firestore/core/src/model/resource_path.h"
 #include "Firestore/core/src/util/hard_assert.h"
+#include "Firestore/third_party/nlohmann_json/json.hpp"
 #include "absl/strings/match.h"
 
 namespace firebase {
 namespace firestore {
 namespace local {
 
+using model::DocumentKey;
+using model::FieldIndex;
+using model::IndexState;
 using model::ResourcePath;
+using model::SnapshotVersion;
+using nlohmann::json;
 
-LevelDbIndexManager::LevelDbIndexManager(LevelDbPersistence* db) : db_(db) {
+namespace {
+
+struct DbIndexState {
+  int64_t seconds;
+  int32_t nanos;
+  std::string key;
+  model::ListenSequenceNumber sequence_number;
+};
+
+// TODO(wuandy): Uncomment this when needed.
+// void to_json(json& j, const DbIndexState& s) {
+//  j = json{{"seconds", s.seconds},
+//           {"nanos", s.nanos},
+//           {"key", s.key},
+//           {"seq_num", s.sequence_number}};
+//}
+
+void from_json(const json& j, DbIndexState& s) {
+  j.at("seconds").get_to(s.seconds);
+  j.at("nanos").get_to(s.nanos);
+  j.at("key").get_to(s.key);
+  j.at("seq_num").get_to(s.sequence_number);
+}
+
+IndexState DecodeIndexState(const std::string& encoded) {
+  auto j = json::parse(encoded.begin(), encoded.end(), /*callback=*/nullptr,
+                       /*allow_exceptions=*/false);
+  auto db_state = j.get<DbIndexState>();
+  return {db_state.sequence_number,
+          SnapshotVersion(Timestamp(db_state.seconds, db_state.nanos)),
+          DocumentKey::FromPathString(db_state.key)};
+}
+
+}  // namespace
+
+LevelDbIndexManager::LevelDbIndexManager(LevelDbPersistence* db,
+                                         LocalSerializer* serializer)
+    : db_(db), serializer_(serializer) {
+  auto cmp = [](FieldIndex* left, FieldIndex* right) {
+    return left->index_state().sequence_number() <
+           right->index_state().sequence_number();
+  };
+  next_index_to_update_ = std::priority_queue<
+      FieldIndex*, std::vector<FieldIndex*>,
+      std::function<bool(model::FieldIndex*, model::FieldIndex*)>>(cmp);
 }
 
 void LevelDbIndexManager::AddToCollectionParentIndex(
@@ -71,6 +125,230 @@ std::vector<ResourcePath> LevelDbIndexManager::GetCollectionParents(
   return results;
 }
 
+void LevelDbIndexManager::Start() {
+  std::unordered_map<int32_t, IndexState> index_states;
+
+  // Fetch all index states that are persisted for the user. These states
+  // contain per user information on how up to date the index is.
+  {
+    auto state_iter = db_->current_transaction()->NewIterator();
+    auto state_key_prefix = LevelDbIndexStateKey::KeyPrefix(uid_);
+    LevelDbIndexStateKey state_key;
+    for (state_iter->Seek(state_key_prefix); state_iter->Valid();
+         state_iter->Next()) {
+      if (!absl::StartsWith(state_iter->key(), state_key_prefix) ||
+          !state_key.Decode(state_iter->key())) {
+        break;
+      }
+
+      index_states.insert(
+          {state_key.index_id(), DecodeIndexState(state_iter->value())});
+    }
+  }
+
+  // Fetch all indices and combine with user's index state if available.
+  {
+    auto config_iter = db_->current_transaction()->NewIterator();
+    auto config_key_prefix = LevelDbIndexConfigurationKey::KeyPrefix();
+    LevelDbIndexConfigurationKey config_key;
+    for (config_iter->Seek(config_key_prefix); config_iter->Valid();
+         config_iter->Next()) {
+      if (!absl::StartsWith(config_iter->key(), config_key_prefix) ||
+          !config_key.Decode(config_iter->key())) {
+        break;
+      }
+
+      nanopb::StringReader reader{config_iter->value()};
+      auto message =
+          nanopb::Message<google_firestore_admin_v1_Index>::TryParse(&reader);
+      if (!reader.ok()) {
+        HARD_FAIL("Index proto failed to parse: %s",
+                  reader.status().ToString());
+      }
+
+      auto segments = serializer_->DecodeFieldIndexSegments(&reader, *message);
+      if (!reader.ok()) {
+        HARD_FAIL("Index proto failed to decode: %s",
+                  reader.status().ToString());
+      }
+
+      // If we fetched an index state for the user above, combine it with this
+      // index. We use the default state if we don't have an index state (e.g.
+      // the index was created while a different user was logged in).
+      auto iter = index_states.find(config_key.index_id());
+      IndexState state = iter != index_states.end()
+                             ? iter->second
+                             : FieldIndex::InitialState();
+
+      // Store the index and update `memoized_max_index_id_` and
+      // `memoized_max_sequence_number_`.
+      MemoizeIndex(FieldIndex(config_key.index_id(),
+                              config_key.collection_group(),
+                              std::move(segments), state));
+    }
+  }
+
+  started_ = true;
+}
+
+void LevelDbIndexManager::DeleteFromUpdateQueue(FieldIndex* index_ptr) {
+  // Pop and save `FieldIndex*` until index_ptr is found, then pushed what are
+  // popped out back to `next_index_to_update_` except for `index_ptr`.
+  std::vector<FieldIndex*> popped_out;
+  while (!next_index_to_update_.empty()) {
+    auto* top = next_index_to_update_.top();
+    next_index_to_update_.pop();
+    if (top == index_ptr) {
+      break;
+    } else {
+      popped_out.push_back(top);
+    }
+  }
+
+  for (auto* index : popped_out) {
+    next_index_to_update_.push(index);
+  }
+}
+
+void LevelDbIndexManager::MemoizeIndex(FieldIndex index) {
+  auto& existing_indexes = memoized_indexes_[index.collection_group()];
+
+  // Copy some value out because `index` will be moved to `existing_index_`
+  // later.
+  auto index_id = index.index_id();
+  auto sequence_number = index.index_state().sequence_number();
+
+  auto existing_index_iter = existing_indexes.find(index_id);
+
+  if (existing_index_iter != existing_indexes.end()) {
+    DeleteFromUpdateQueue(&existing_index_iter->second);
+  }
+
+  // Moves `index` into `existing_indexes`.
+  existing_indexes.insert({index_id, std::move(index)});
+  // next_index_to_update_ holds a pointer to the index owned by
+  // `existing_indexes`.
+  next_index_to_update_.push(&existing_indexes.find(index_id)->second);
+  memoized_max_index_id_ = std::max(memoized_max_index_id_, index_id);
+  memoized_max_sequence_number_ =
+      std::max(memoized_max_sequence_number_, sequence_number);
+}
+
+void LevelDbIndexManager::AddFieldIndex(const FieldIndex& index) {
+  HARD_ASSERT(started_, "IndexManager not started");
+
+  int next_index_id = memoized_max_index_id_ + 1;
+  FieldIndex new_index(next_index_id, index.collection_group(),
+                       index.segments(), index.index_state());
+
+  auto config_key = LevelDbIndexConfigurationKey::Key(
+      new_index.index_id(), new_index.collection_group());
+  db_->current_transaction()->Put(
+      config_key, serializer_->EncodeFieldIndexSegments(new_index.segments()));
+
+  MemoizeIndex(std::move(new_index));
+}
+
+void LevelDbIndexManager::DeleteFieldIndex(const FieldIndex& index) {
+  HARD_ASSERT(started_, "IndexManager not started");
+
+  db_->current_transaction()->Delete(LevelDbIndexConfigurationKey::Key(
+      index.index_id(), index.collection_group()));
+
+  // Delete states from all users for this index id.
+  {
+    auto state_prefix = LevelDbIndexStateKey::KeyPrefix();
+    auto iter = db_->current_transaction()->NewIterator();
+    LevelDbIndexStateKey state_key;
+    for (iter->Seek(state_prefix); iter->Valid(); iter->Next()) {
+      if (!absl::StartsWith(iter->key(), state_prefix) ||
+          !state_key.Decode(iter->key())) {
+        break;
+      }
+
+      if (state_key.index_id() == index.index_id()) {
+        db_->current_transaction()->Delete(iter->key());
+      }
+    }
+  }
+
+  // Delete entries from all users for this index id.
+  {
+    auto entry_prefix = LevelDbIndexEntryKey::KeyPrefix(index.index_id());
+    auto iter = db_->current_transaction()->NewIterator();
+    for (iter->Seek(entry_prefix); iter->Valid(); iter->Next()) {
+      if (!absl::StartsWith(iter->key(), entry_prefix)) {
+        break;
+      }
+      db_->current_transaction()->Delete(iter->key());
+    }
+  }
+
+  auto group_index_iter = memoized_indexes_.find(index.collection_group());
+  if (group_index_iter != memoized_indexes_.end()) {
+    auto& index_map = group_index_iter->second;
+    auto index_iter = index_map.find(index.index_id());
+    if (index_iter != index_map.end()) {
+      DeleteFromUpdateQueue(&index_iter->second);
+      index_map.erase(index_iter);
+    }
+  }
+}
+
+std::vector<FieldIndex> LevelDbIndexManager::GetFieldIndexes(
+    const std::string& collection_group) {
+  HARD_ASSERT(started_, "IndexManager not started");
+
+  std::vector<FieldIndex> result;
+  const auto& indexes = memoized_indexes_[collection_group];
+  for (const auto& entry : indexes) {
+    result.push_back(entry.second);
+  }
+
+  return result;
+}
+
+std::vector<model::FieldIndex> LevelDbIndexManager::GetFieldIndexes() {
+  std::vector<FieldIndex> result;
+  for (const auto& entry : memoized_indexes_) {
+    for (const auto& id_index_entry : entry.second) {
+      result.push_back(id_index_entry.second);
+    }
+  }
+
+  return result;
+}
+
+absl::optional<model::FieldIndex> LevelDbIndexManager::GetFieldIndex(
+    core::Target target) {
+  (void)target;
+  return {};
+}
+
+absl::optional<std::vector<model::DocumentKey>>
+LevelDbIndexManager::GetDocumentsMatchingTarget(model::FieldIndex field_index,
+                                                core::Target target) {
+  (void)field_index;
+  (void)target;
+  return {};
+}
+
+absl::optional<std::string>
+LevelDbIndexManager::GetNextCollectionGroupToUpdate() {
+  return {};
+}
+
+void LevelDbIndexManager::UpdateCollectionGroup(
+    const std::string& collection_group, model::IndexOffset offset) {
+  (void)collection_group;
+  (void)offset;
+}
+
+void LevelDbIndexManager::UpdateIndexEntries(
+    const model::DocumentMap& documents) {
+  (void)documents;
+}
+
 }  // namespace local
 }  // namespace firestore
 }  // namespace firebase
diff --git a/Firestore/core/src/local/leveldb_index_manager.h b/Firestore/core/src/local/leveldb_index_manager.h
index 233b8dd58b8..9af99a3a905 100644
--- a/Firestore/core/src/local/leveldb_index_manager.h
+++ b/Firestore/core/src/local/leveldb_index_manager.h
@@ -17,22 +17,29 @@
 #ifndef FIRESTORE_CORE_SRC_LOCAL_LEVELDB_INDEX_MANAGER_H_
 #define FIRESTORE_CORE_SRC_LOCAL_LEVELDB_INDEX_MANAGER_H_
 
+#include <queue>
 #include <string>
+#include <unordered_map>
 #include <vector>
 
 #include "Firestore/core/src/local/index_manager.h"
 #include "Firestore/core/src/local/memory_index_manager.h"
+#include "Firestore/core/src/model/field_index.h"
 
 namespace firebase {
 namespace firestore {
 namespace local {
 
 class LevelDbPersistence;
+class LocalSerializer;
 
 /** A persisted implementation of IndexManager. */
 class LevelDbIndexManager : public IndexManager {
  public:
-  explicit LevelDbIndexManager(LevelDbPersistence* db);
+  explicit LevelDbIndexManager(LevelDbPersistence* db,
+                               LocalSerializer* serializer);
+
+  void Start() override;
 
   void AddToCollectionParentIndex(
       const model::ResourcePath& collection_path) override;
@@ -40,7 +47,42 @@ class LevelDbIndexManager : public IndexManager {
   std::vector<model::ResourcePath> GetCollectionParents(
       const std::string& collection_id) override;
 
+  void AddFieldIndex(const model::FieldIndex& index) override;
+
+  void DeleteFieldIndex(const model::FieldIndex& index) override;
+
+  std::vector<model::FieldIndex> GetFieldIndexes(
+      const std::string& collection_group) override;
+
+  std::vector<model::FieldIndex> GetFieldIndexes() override;
+
+  absl::optional<model::FieldIndex> GetFieldIndex(core::Target target) override;
+
+  absl::optional<std::vector<model::DocumentKey>> GetDocumentsMatchingTarget(
+      model::FieldIndex fieldIndex, core::Target target) override;
+
+  absl::optional<std::string> GetNextCollectionGroupToUpdate() override;
+
+  void UpdateCollectionGroup(const std::string& collection_group,
+                             model::IndexOffset offset) override;
+
+  void UpdateIndexEntries(const model::DocumentMap& documents) override;
+
  private:
+  using QueueForNextIndexToUpdate = std::priority_queue<
+      model::FieldIndex*,
+      std::vector<model::FieldIndex*>,
+      std::function<bool(model::FieldIndex*, model::FieldIndex*)>>;
+
+  /**
+   * Stores the index in the memoized indexes table and updates
+   * `next_index_to_update_` `memoized_max_index_id_` and
+   * `memoized_max_sequence_number_`.
+   */
+  void MemoizeIndex(model::FieldIndex index);
+
+  void DeleteFromUpdateQueue(model::FieldIndex* index);
+
   // The LevelDbIndexManager is owned by LevelDbPersistence.
   LevelDbPersistence* db_;
 
@@ -52,6 +94,27 @@ class LevelDbIndexManager : public IndexManager {
    * be used to satisfy reads.
    */
   MemoryCollectionParentIndex collection_parents_cache_;
+
+  /**
+   * An in-memory map from collection group to a map of indexes associated with
+   * the collection groups.
+   *
+   * The nested map is an index_id to FieldIndex map.
+   */
+  std::unordered_map<std::string,
+                     std::unordered_map<int32_t, model::FieldIndex>>
+      memoized_indexes_;
+
+  QueueForNextIndexToUpdate next_index_to_update_;
+  int32_t memoized_max_index_id_ = -1;
+  int64_t memoized_max_sequence_number_ = -1;
+
+  /* Owned by LevelDbPersistence. */
+  LocalSerializer* serializer_ = nullptr;
+
+  bool started_ = false;
+
+  std::string uid_;
 };
 
 }  // namespace local
diff --git a/Firestore/core/src/local/leveldb_key.cc b/Firestore/core/src/local/leveldb_key.cc
index 8f13f09a3f5..11334138703 100644
--- a/Firestore/core/src/local/leveldb_key.cc
+++ b/Firestore/core/src/local/leveldb_key.cc
@@ -127,6 +127,9 @@ enum ComponentLabel {
   /** A component containing a directional index value. */
   IndexDirectionalValue = 21,
 
+  /** A component containing a collection group name. */
+  CollectionGroup = 22,
+
   /**
    * A path segment describes just a single segment in a resource path. Path
    * segments that occur sequentially in a key represent successive segments in
@@ -217,6 +220,10 @@ class Reader {
     return ReadLabeledInt32(ComponentLabel::IndexId);
   }
 
+  std::string ReadCollectionGroup() {
+    return ReadLabeledString(ComponentLabel::CollectionGroup);
+  }
+
   std::string ReadIndexArrayValue() {
     return ReadLabeledString(ComponentLabel::IndexArrayValue);
   }
@@ -596,6 +603,11 @@ std::string Reader::Describe() {
       if (ok_) {
         absl::StrAppend(&description, " index_id=", index_id);
       }
+    } else if (label == ComponentLabel::CollectionGroup) {
+      auto group = ReadCollectionGroup();
+      if (ok_) {
+        absl::StrAppend(&description, " collection_group=", group);
+      }
     } else if (label == ComponentLabel::IndexArrayValue) {
       std::string value = ReadIndexArrayValue();
       if (ok_) {
@@ -693,6 +705,10 @@ class Writer {
     WriteLabeledInt32(ComponentLabel::IndexId, id);
   }
 
+  void WriteCollectionGroup(absl::string_view collection_group) {
+    WriteLabeledString(ComponentLabel::CollectionGroup, collection_group);
+  }
+
   void WriteIndexArrayValue(absl::string_view value) {
     WriteLabeledString(ComponentLabel::IndexArrayValue, value);
   }
@@ -1151,10 +1167,12 @@ std::string LevelDbIndexConfigurationKey::KeyPrefix() {
   return writer.result();
 }
 
-std::string LevelDbIndexConfigurationKey::Key(int32_t id) {
+std::string LevelDbIndexConfigurationKey::Key(
+    int32_t id, absl::string_view collection_group) {
   Writer writer;
   writer.WriteTableName(kIndexConfigurationTable);
   writer.WriteIndexId(id);
+  writer.WriteCollectionGroup(collection_group);
   writer.WriteTerminator();
   return writer.result();
 }
@@ -1163,6 +1181,7 @@ bool LevelDbIndexConfigurationKey::Decode(absl::string_view key) {
   Reader reader{key};
   reader.ReadTableNameMatching(kIndexConfigurationTable);
   index_id_ = reader.ReadIndexId();
+  collection_group_ = reader.ReadCollectionGroup();
   reader.ReadTerminator();
   return reader.ok();
 }
@@ -1173,12 +1192,19 @@ std::string LevelDbIndexStateKey::KeyPrefix() {
   return writer.result();
 }
 
-std::string LevelDbIndexStateKey::Key(int32_t index_id,
-                                      absl::string_view user_id) {
+std::string LevelDbIndexStateKey::KeyPrefix(absl::string_view user_id) {
   Writer writer;
   writer.WriteTableName(kIndexStateTable);
-  writer.WriteIndexId(index_id);
   writer.WriteUserId(user_id);
+  return writer.result();
+}
+
+std::string LevelDbIndexStateKey::Key(absl::string_view user_id,
+                                      int32_t index_id) {
+  Writer writer;
+  writer.WriteTableName(kIndexStateTable);
+  writer.WriteUserId(user_id);
+  writer.WriteIndexId(index_id);
   writer.WriteTerminator();
   return writer.result();
 }
@@ -1186,8 +1212,8 @@ std::string LevelDbIndexStateKey::Key(int32_t index_id,
 bool LevelDbIndexStateKey::Decode(absl::string_view key) {
   Reader reader{key};
   reader.ReadTableNameMatching(kIndexStateTable);
-  index_id_ = reader.ReadIndexId();
   user_id_ = reader.ReadUserId();
+  index_id_ = reader.ReadIndexId();
   reader.ReadTerminator();
   return reader.ok();
 }
@@ -1198,6 +1224,13 @@ std::string LevelDbIndexEntryKey::KeyPrefix() {
   return writer.result();
 }
 
+std::string LevelDbIndexEntryKey::KeyPrefix(int32_t index_id) {
+  Writer writer;
+  writer.WriteTableName(kIndexEntriesTable);
+  writer.WriteIndexId(index_id);
+  return writer.result();
+}
+
 std::string LevelDbIndexEntryKey::Key(int32_t index_id,
                                       absl::string_view user_id,
                                       absl::string_view array_value,
diff --git a/Firestore/core/src/local/leveldb_key.h b/Firestore/core/src/local/leveldb_key.h
index be071e859cc..509d7ed1ab0 100644
--- a/Firestore/core/src/local/leveldb_key.h
+++ b/Firestore/core/src/local/leveldb_key.h
@@ -744,7 +744,7 @@ class LevelDbIndexConfigurationKey {
   /**
    * Creates a key that points to the key for the given index id.
    */
-  static std::string Key(int32_t id);
+  static std::string Key(int32_t id, absl::string_view collection_group);
 
   /**
    * Decodes the given complete key, storing the decoded values in this
@@ -762,8 +762,14 @@ class LevelDbIndexConfigurationKey {
     return index_id_;
   }
 
+  /** The collection group for this index. */
+  const std::string& collection_group() const {
+    return collection_group_;
+  }
+
  private:
   int32_t index_id_;
+  std::string collection_group_;
 };
 
 /**
@@ -778,9 +784,15 @@ class LevelDbIndexStateKey {
   static std::string KeyPrefix();
 
   /**
-   * Creates a key that points to the key for the given index id and user id.
+   * Creates a key prefix that points just before the first key for a given user
+   * id.
+   */
+  static std::string KeyPrefix(absl::string_view user_id);
+
+  /**
+   * Creates a key that points to the key for the given user id and index id.
    */
-  static std::string Key(int32_t index_id, absl::string_view user_id);
+  static std::string Key(absl::string_view user_id, int32_t index_id);
 
   /**
    * Decodes the given complete key, storing the decoded values in this
@@ -793,16 +805,16 @@ class LevelDbIndexStateKey {
   ABSL_MUST_USE_RESULT
   bool Decode(absl::string_view key);
 
-  /** The index id for this entry. */
-  int32_t index_id() const {
-    return index_id_;
-  }
-
   /** The user id for this entry. */
   const std::string& user_id() const {
     return user_id_;
   }
 
+  /** The index id for this entry. */
+  int32_t index_id() const {
+    return index_id_;
+  }
+
  private:
   int32_t index_id_;
   std::string user_id_;
@@ -821,6 +833,11 @@ class LevelDbIndexEntryKey {
    */
   static std::string KeyPrefix();
 
+  /**
+   * Creates a key prefix that points the first entry of a given index_id.
+   */
+  static std::string KeyPrefix(int32_t index_id);
+
   /**
    * Creates a key that points to the key for the given index entry fields.
    */
diff --git a/Firestore/core/src/local/leveldb_persistence.cc b/Firestore/core/src/local/leveldb_persistence.cc
index 6d188c98984..4715655ff80 100644
--- a/Firestore/core/src/local/leveldb_persistence.cc
+++ b/Firestore/core/src/local/leveldb_persistence.cc
@@ -114,7 +114,7 @@ LevelDbPersistence::LevelDbPersistence(std::unique_ptr<leveldb::DB> db,
   target_cache_ = absl::make_unique<LevelDbTargetCache>(this, &serializer_);
   document_cache_ =
       absl::make_unique<LevelDbRemoteDocumentCache>(this, &serializer_);
-  index_manager_ = absl::make_unique<LevelDbIndexManager>(this);
+  index_manager_ = absl::make_unique<LevelDbIndexManager>(this, &serializer_);
   reference_delegate_ =
       absl::make_unique<LevelDbLruReferenceDelegate>(this, lru_params);
   bundle_cache_ = absl::make_unique<LevelDbBundleCache>(this, &serializer_);
diff --git a/Firestore/core/src/local/local_serializer.cc b/Firestore/core/src/local/local_serializer.cc
index 770c5d9474b..a058235a236 100644
--- a/Firestore/core/src/local/local_serializer.cc
+++ b/Firestore/core/src/local/local_serializer.cc
@@ -26,11 +26,13 @@
 #include "Firestore/Protos/nanopb/firestore/local/maybe_document.nanopb.h"
 #include "Firestore/Protos/nanopb/firestore/local/mutation.nanopb.h"
 #include "Firestore/Protos/nanopb/firestore/local/target.nanopb.h"
+#include "Firestore/Protos/nanopb/google/firestore/admin/index.nanopb.h"
 #include "Firestore/Protos/nanopb/google/firestore/v1/document.nanopb.h"
 #include "Firestore/core/src/bundle/bundle_metadata.h"
 #include "Firestore/core/src/bundle/named_query.h"
 #include "Firestore/core/src/core/query.h"
 #include "Firestore/core/src/local/target_data.h"
+#include "Firestore/core/src/model/field_path.h"
 #include "Firestore/core/src/model/mutable_document.h"
 #include "Firestore/core/src/model/mutation_batch.h"
 #include "Firestore/core/src/model/snapshot_version.h"
@@ -38,6 +40,7 @@
 #include "Firestore/core/src/nanopb/message.h"
 #include "Firestore/core/src/nanopb/nanopb_util.h"
 #include "Firestore/core/src/util/hard_assert.h"
+#include "Firestore/core/src/util/statusor.h"
 #include "Firestore/core/src/util/string_format.h"
 #include "absl/types/span.h"
 
@@ -51,11 +54,13 @@ using bundle::BundleMetadata;
 using bundle::NamedQuery;
 using core::Target;
 using model::DeepClone;
+using model::FieldPath;
 using model::FieldTransform;
 using model::MutableDocument;
 using model::Mutation;
 using model::MutationBatch;
 using model::ObjectValue;
+using model::Segment;
 using model::SnapshotVersion;
 using nanopb::ByteString;
 using nanopb::CheckedSize;
@@ -453,6 +458,87 @@ BundledQuery LocalSerializer::DecodeBundledQuery(
       limit_type);
 }
 
+std::vector<model::Segment> LocalSerializer::DecodeFieldIndexSegments(
+    nanopb::Reader* reader, google_firestore_admin_v1_Index& index) const {
+  std::vector<model::Segment> result;
+  for (size_t i = 0; i < index.fields_count; ++i) {
+    const auto& field = index.fields[i];
+
+    util::StatusOr<FieldPath> field_path =
+        FieldPath::FromServerFormat(nanopb::MakeString(field.field_path));
+    if (!field_path.ok()) {
+      reader->Fail(
+          StringFormat("Failed to read field path for index segment: %s",
+                       nanopb::MakeString(field.field_path)));
+      return {};
+    }
+
+    Segment::Kind kind = Segment::Kind::kContains;
+    if (field.which_value_mode ==
+        google_firestore_admin_v1_Index_IndexField_order_tag) {
+      if (field.order ==
+          google_firestore_admin_v1_Index_IndexField_Order_ASCENDING) {
+        kind = Segment::Kind::kAscending;
+      } else {
+        kind = Segment::Kind::kDescending;
+      }
+    }
+
+    result.push_back({field_path.ValueOrDie(), kind});
+  }
+
+  return result;
+}
+
+nanopb::Message<google_firestore_admin_v1_Index>
+LocalSerializer::EncodeFieldIndexSegments(
+    const std::vector<model::Segment>& segments) const {
+  Message<google_firestore_admin_v1_Index> result;
+
+  result->query_scope =
+      google_firestore_admin_v1_Index_QueryScope_COLLECTION_GROUP;
+
+  result->fields_count = segments.size();
+  result->fields =
+      MakeArray<google_firestore_admin_v1_Index_IndexField>(segments.size());
+  int i = 0;
+  for (const auto& segment : segments) {
+    google_firestore_admin_v1_Index_IndexField field;
+    field.field_path =
+        nanopb::MakeBytesArray(segment.field_path().CanonicalString());
+    switch (segment.kind()) {
+      case model::Segment::kContains: {
+        field.which_value_mode =
+            google_firestore_admin_v1_Index_IndexField_array_config_tag;
+        field.array_config =
+            google_firestore_admin_v1_Index_IndexField_ArrayConfig_CONTAINS;
+        break;
+      }
+      case model::Segment::kAscending: {
+        field.which_value_mode =
+            google_firestore_admin_v1_Index_IndexField_order_tag;
+        field.order =
+            google_firestore_admin_v1_Index_IndexField_Order_ASCENDING;
+        break;
+      }
+      case model::Segment::kDescending: {
+        field.which_value_mode =
+            google_firestore_admin_v1_Index_IndexField_order_tag;
+        field.order =
+            google_firestore_admin_v1_Index_IndexField_Order_DESCENDING;
+        break;
+      }
+      default:
+        HARD_FAIL("Unrecognized enum value from segment.kind()");
+    }
+
+    result->fields[i] = std::move(field);
+    ++i;
+  }
+
+  return result;
+}
+
 }  // namespace local
 }  // namespace firestore
 }  // namespace firebase
diff --git a/Firestore/core/src/local/local_serializer.h b/Firestore/core/src/local/local_serializer.h
index 2a7396ec685..d129ffe5d32 100644
--- a/Firestore/core/src/local/local_serializer.h
+++ b/Firestore/core/src/local/local_serializer.h
@@ -21,6 +21,7 @@
 #include <utility>
 #include <vector>
 
+#include "Firestore/core/src/model/field_index.h"
 #include "Firestore/core/src/model/model_fwd.h"
 #include "Firestore/core/src/model/types.h"
 #include "Firestore/core/src/remote/serializer.h"
@@ -38,6 +39,7 @@ typedef struct _firestore_client_Target firestore_client_Target;
 typedef struct _firestore_client_UnknownDocument
     firestore_client_UnknownDocument;
 typedef struct _firestore_client_WriteBatch firestore_client_WriteBatch;
+typedef struct _google_firestore_admin_v1_Index google_firestore_admin_v1_Index;
 
 namespace nanopb {
 template <typename T>
@@ -148,6 +150,12 @@ class LocalSerializer {
   bundle::NamedQuery DecodeNamedQuery(nanopb::Reader* reader,
                                       firestore_NamedQuery& proto) const;
 
+  nanopb::Message<google_firestore_admin_v1_Index> EncodeFieldIndexSegments(
+      const std::vector<model::Segment>& segments) const;
+
+  std::vector<model::Segment> DecodeFieldIndexSegments(
+      nanopb::Reader* reader, google_firestore_admin_v1_Index& index) const;
+
  private:
   /**
    * Encodes a Document for local storage. This differs from the v1 RPC
diff --git a/Firestore/core/src/local/memory_index_manager.cc b/Firestore/core/src/local/memory_index_manager.cc
index d8ec923249c..9fdc609c690 100644
--- a/Firestore/core/src/local/memory_index_manager.cc
+++ b/Firestore/core/src/local/memory_index_manager.cc
@@ -21,6 +21,9 @@
 #include <unordered_map>
 #include <vector>
 
+#include "Firestore/core/src/core/target.h"
+#include "Firestore/core/src/model/field_index.h"
+#include "Firestore/core/src/model/model_fwd.h"
 #include "Firestore/core/src/model/resource_path.h"
 #include "Firestore/core/src/util/hard_assert.h"
 
@@ -62,6 +65,60 @@ std::vector<ResourcePath> MemoryIndexManager::GetCollectionParents(
   return collection_parents_index_.GetEntries(collection_id);
 }
 
+// Below methods are only stubs because field indices are not supported with
+// memory persistence.
+
+void MemoryIndexManager::Start() {
+}
+
+void MemoryIndexManager::AddFieldIndex(const model::FieldIndex& index) {
+  (void)index;
+}
+
+void MemoryIndexManager::DeleteFieldIndex(const model::FieldIndex& index) {
+  (void)index;
+}
+
+std::vector<model::FieldIndex> MemoryIndexManager::GetFieldIndexes(
+    const std::string& collection_group) {
+  (void)collection_group;
+  return {};
+}
+
+std::vector<model::FieldIndex> MemoryIndexManager::GetFieldIndexes() {
+  return {};
+}
+
+absl::optional<model::FieldIndex> MemoryIndexManager::GetFieldIndex(
+    core::Target target) {
+  (void)target;
+  return absl::nullopt;
+}
+
+absl::optional<std::vector<model::DocumentKey>>
+MemoryIndexManager::GetDocumentsMatchingTarget(model::FieldIndex fieldIndex,
+                                               core::Target target) {
+  (void)fieldIndex;
+  (void)target;
+  return {};
+}
+
+absl::optional<std::string>
+MemoryIndexManager::GetNextCollectionGroupToUpdate() {
+  return absl::nullopt;
+}
+
+void MemoryIndexManager::UpdateCollectionGroup(
+    const std::string& collection_group, model::IndexOffset offset) {
+  (void)collection_group;
+  (void)offset;
+}
+
+void MemoryIndexManager::UpdateIndexEntries(
+    const model::DocumentMap& documents) {
+  (void)documents;
+}
+
 }  // namespace local
 }  // namespace firestore
 }  // namespace firebase
diff --git a/Firestore/core/src/local/memory_index_manager.h b/Firestore/core/src/local/memory_index_manager.h
index f3f7bef0b23..bd2afcdc129 100644
--- a/Firestore/core/src/local/memory_index_manager.h
+++ b/Firestore/core/src/local/memory_index_manager.h
@@ -48,12 +48,37 @@ class MemoryCollectionParentIndex {
 /** An in-memory implementation of IndexManager. */
 class MemoryIndexManager : public IndexManager {
  public:
+  MemoryIndexManager() = default;
+
+  void Start() override;
+
   void AddToCollectionParentIndex(
       const model::ResourcePath& collection_path) override;
 
   std::vector<model::ResourcePath> GetCollectionParents(
       const std::string& collection_id) override;
 
+  void AddFieldIndex(const model::FieldIndex& index) override;
+
+  void DeleteFieldIndex(const model::FieldIndex& index) override;
+
+  std::vector<model::FieldIndex> GetFieldIndexes(
+      const std::string& collection_group) override;
+
+  std::vector<model::FieldIndex> GetFieldIndexes() override;
+
+  absl::optional<model::FieldIndex> GetFieldIndex(core::Target target) override;
+
+  absl::optional<std::vector<model::DocumentKey>> GetDocumentsMatchingTarget(
+      model::FieldIndex fieldIndex, core::Target target) override;
+
+  absl::optional<std::string> GetNextCollectionGroupToUpdate() override;
+
+  void UpdateCollectionGroup(const std::string& collection_group,
+                             model::IndexOffset offset) override;
+
+  void UpdateIndexEntries(const model::DocumentMap& documents) override;
+
  private:
   MemoryCollectionParentIndex collection_parents_index_;
 };
diff --git a/Firestore/core/src/nanopb/fields_array.h b/Firestore/core/src/nanopb/fields_array.h
index 15849d4d0ab..9c8c45c7ebf 100644
--- a/Firestore/core/src/nanopb/fields_array.h
+++ b/Firestore/core/src/nanopb/fields_array.h
@@ -21,6 +21,7 @@
 #include "Firestore/Protos/nanopb/firestore/local/maybe_document.nanopb.h"
 #include "Firestore/Protos/nanopb/firestore/local/mutation.nanopb.h"
 #include "Firestore/Protos/nanopb/firestore/local/target.nanopb.h"
+#include "Firestore/Protos/nanopb/google/firestore/admin/index.nanopb.h"
 #include "Firestore/Protos/nanopb/google/firestore/v1/document.nanopb.h"
 #include "Firestore/Protos/nanopb/google/firestore/v1/firestore.nanopb.h"
 #include "Firestore/Protos/nanopb/google/type/latlng.nanopb.h"
@@ -179,6 +180,11 @@ inline const pb_field_t* FieldsArray<firestore_NamedQuery>() {
   return firestore_NamedQuery_fields;
 }
 
+template <>
+inline const pb_field_t* FieldsArray<google_firestore_admin_v1_Index>() {
+  return google_protobuf_Empty_fields;
+}
+
 template <>
 inline const pb_field_t* FieldsArray<google_protobuf_Empty>() {
   return google_protobuf_Empty_fields;
diff --git a/Firestore/core/test/unit/local/leveldb_index_manager_test.cc b/Firestore/core/test/unit/local/leveldb_index_manager_test.cc
index 2a0ca188a02..4e4c3324490 100644
--- a/Firestore/core/test/unit/local/leveldb_index_manager_test.cc
+++ b/Firestore/core/test/unit/local/leveldb_index_manager_test.cc
@@ -19,6 +19,7 @@
 #include "Firestore/core/src/local/leveldb_index_manager.h"
 #include "Firestore/core/src/local/leveldb_persistence.h"
 #include "Firestore/core/test/unit/local/persistence_testing.h"
+#include "Firestore/core/test/unit/testutil/testutil.h"
 #include "absl/memory/memory.h"
 #include "gtest/gtest.h"
 
@@ -28,6 +29,11 @@ namespace local {
 
 namespace {
 
+using model::FieldIndex;
+using model::ResourcePath;
+using model::Segment;
+using testutil::MakeFieldIndex;
+
 std::unique_ptr<Persistence> PersistenceFactory() {
   return LevelDbPersistenceForTesting();
 }
@@ -38,6 +44,75 @@ INSTANTIATE_TEST_SUITE_P(LevelDbIndexManagerTest,
                          IndexManagerTest,
                          ::testing::Values(PersistenceFactory));
 
+class LevelDbIndexManagerTest : public ::testing::Test {
+ public:
+  // `GetParam()` must return a factory function.
+  LevelDbIndexManagerTest() : persistence{PersistenceFactory()} {
+  }
+
+  std::unique_ptr<Persistence> persistence;
+};
+
+TEST_F(LevelDbIndexManagerTest, CreateReadFieldsIndexes) {
+  persistence->Run("CreateReadDeleteFieldsIndexes", [&]() {
+    IndexManager* index_manager = persistence->index_manager();
+    index_manager->Start();
+
+    index_manager->AddFieldIndex(
+        MakeFieldIndex("coll1", 1, model::FieldIndex::InitialState(), "value",
+                       model::Segment::kAscending));
+    index_manager->AddFieldIndex(
+        MakeFieldIndex("coll2", 2, model::FieldIndex::InitialState(), "value",
+                       model::Segment::kContains));
+
+    {
+      auto indexes = index_manager->GetFieldIndexes("coll1");
+      EXPECT_EQ(indexes.size(), 1);
+      // Note index_id() is 0 because index manager rewrites it using its
+      // internal id.
+      EXPECT_EQ(indexes[0].index_id(), 0);
+      EXPECT_EQ(indexes[0].collection_group(), "coll1");
+    }
+
+    index_manager->AddFieldIndex(
+        MakeFieldIndex("coll1", 3, model::FieldIndex::InitialState(),
+                       "newValue", model::Segment::kContains));
+    {
+      auto indexes = index_manager->GetFieldIndexes("coll1");
+      EXPECT_EQ(indexes.size(), 2);
+      EXPECT_EQ(indexes[0].collection_group(), "coll1");
+      EXPECT_EQ(indexes[1].collection_group(), "coll1");
+    }
+
+    {
+      auto indexes = index_manager->GetFieldIndexes("coll2");
+      EXPECT_EQ(indexes.size(), 1);
+      EXPECT_EQ(indexes[0].collection_group(), "coll2");
+    }
+  });
+}
+
+TEST_F(LevelDbIndexManagerTest, DeleteFieldsIndexeRemovesAllMetadata) {
+  persistence->Run("CreateReadDeleteFieldsIndexes", [&]() {
+    IndexManager* index_manager = persistence->index_manager();
+    index_manager->Start();
+
+    auto index = MakeFieldIndex("coll1", 0, model::FieldIndex::InitialState(),
+                                "value", model::Segment::kAscending);
+    index_manager->AddFieldIndex(index);
+    {
+      auto indexes = index_manager->GetFieldIndexes("coll1");
+      EXPECT_EQ(indexes.size(), 1);
+    }
+
+    index_manager->DeleteFieldIndex(index);
+    {
+      auto indexes = index_manager->GetFieldIndexes("coll1");
+      EXPECT_EQ(indexes.size(), 0);
+    }
+  });
+}
+
 }  // namespace local
 }  // namespace firestore
 }  // namespace firebase
diff --git a/Firestore/core/test/unit/local/leveldb_key_test.cc b/Firestore/core/test/unit/local/leveldb_key_test.cc
index 21494ca4983..963d72f1d7c 100644
--- a/Firestore/core/test/unit/local/leveldb_key_test.cc
+++ b/Firestore/core/test/unit/local/leveldb_key_test.cc
@@ -16,6 +16,7 @@
 
 #include "Firestore/core/src/local/leveldb_key.h"
 
+#include "Firestore/core/src/util/autoid.h"
 #include "Firestore/core/src/util/string_util.h"
 #include "Firestore/core/test/unit/testutil/testutil.h"
 #include "absl/strings/match.h"
@@ -494,72 +495,89 @@ TEST(IndexConfigurationKeyTest, Prefixing) {
   auto table_key = LevelDbIndexConfigurationKey::KeyPrefix();
 
   ASSERT_TRUE(
-      absl::StartsWith(LevelDbIndexConfigurationKey::Key(0), table_key));
+      absl::StartsWith(LevelDbIndexConfigurationKey::Key(0, ""), table_key));
 
-  ASSERT_FALSE(absl::StartsWith(LevelDbIndexConfigurationKey::Key(1),
-                                LevelDbIndexConfigurationKey::Key(2)));
+  ASSERT_FALSE(absl::StartsWith(LevelDbIndexConfigurationKey::Key(1, ""),
+                                LevelDbIndexConfigurationKey::Key(2, "")));
+
+  ASSERT_FALSE(absl::StartsWith(LevelDbIndexConfigurationKey::Key(1, "g"),
+                                LevelDbIndexConfigurationKey::Key(1, "ag")));
 }
 
 TEST(IndexConfigurationKeyTest, Ordering) {
-  ASSERT_LT(LevelDbIndexConfigurationKey::Key(0),
-            LevelDbIndexConfigurationKey::Key(1));
-  ASSERT_EQ(LevelDbIndexConfigurationKey::Key(1),
-            LevelDbIndexConfigurationKey::Key(1));
+  ASSERT_LT(LevelDbIndexConfigurationKey::Key(0, ""),
+            LevelDbIndexConfigurationKey::Key(1, ""));
+  ASSERT_EQ(LevelDbIndexConfigurationKey::Key(1, ""),
+            LevelDbIndexConfigurationKey::Key(1, ""));
+  ASSERT_LT(LevelDbIndexConfigurationKey::Key(0, "a"),
+            LevelDbIndexConfigurationKey::Key(0, "b"));
+  ASSERT_EQ(LevelDbIndexConfigurationKey::Key(1, "a"),
+            LevelDbIndexConfigurationKey::Key(1, "a"));
 }
 
 TEST(IndexConfigurationKeyTest, EncodeDecodeCycle) {
   LevelDbIndexConfigurationKey key;
 
+  std::vector<std::string> groups = {
+      "",
+      "ab",
+      "12",
+      ",867t-b",
+      "汉语; traditional Chinese: 漢語; pinyin: Hànyǔ[b]",
+      "اَلْعَرَبِيَّةُ, al-ʿarabiyyah "};
   for (int32_t id = -5; id < 10; ++id) {
-    auto encoded = LevelDbIndexConfigurationKey::Key(id);
+    auto s = groups[(id + 5) % groups.size()];
+    auto encoded = LevelDbIndexConfigurationKey::Key(id, s);
     bool ok = key.Decode(encoded);
     ASSERT_TRUE(ok);
     ASSERT_EQ(id, key.index_id());
+    ASSERT_EQ(s, key.collection_group());
   }
 }
 
 TEST(IndexConfigurationKeyTest, Description) {
-  AssertExpectedKeyDescription("[index_configuration: index_id=8]",
-                               LevelDbIndexConfigurationKey::Key(8));
+  AssertExpectedKeyDescription(
+      "[index_configuration: index_id=8 collection_group=group]",
+      LevelDbIndexConfigurationKey::Key(8, "group"));
 }
 
 TEST(IndexStateKeyTest, Prefixing) {
   auto table_key = LevelDbIndexStateKey::KeyPrefix();
 
   ASSERT_TRUE(
-      absl::StartsWith(LevelDbIndexStateKey::Key(0, "user_a"), table_key));
+      absl::StartsWith(LevelDbIndexStateKey::Key("user_a", 0), table_key));
 
-  ASSERT_FALSE(absl::StartsWith(LevelDbIndexStateKey::Key(0, "user_a"),
-                                LevelDbIndexStateKey::Key(0, "user_b")));
-  ASSERT_FALSE(absl::StartsWith(LevelDbIndexStateKey::Key(0, "user_a"),
-                                LevelDbIndexStateKey::Key(1, "user_a")));
+  ASSERT_FALSE(absl::StartsWith(LevelDbIndexStateKey::Key("user_a", 0),
+                                LevelDbIndexStateKey::Key("user_b", 0)));
+  ASSERT_FALSE(absl::StartsWith(LevelDbIndexStateKey::Key("user_a", 0),
+                                LevelDbIndexStateKey::Key("user_a", 1)));
 }
 
 TEST(IndexStateKeyTest, Ordering) {
-  ASSERT_LT(LevelDbIndexStateKey::Key(0, "foo/bar"),
-            LevelDbIndexStateKey::Key(1, "foo/bar"));
-  ASSERT_LT(LevelDbIndexStateKey::Key(0, "foo/bar"),
-            LevelDbIndexStateKey::Key(0, "foo/bar1"));
+  ASSERT_LT(LevelDbIndexStateKey::Key("foo/bar", 0),
+            LevelDbIndexStateKey::Key("foo/bar", 1));
+  ASSERT_LT(LevelDbIndexStateKey::Key("foo/bar", 0),
+            LevelDbIndexStateKey::Key("foo/bar1", 0));
 }
 
 TEST(IndexStateKeyTest, EncodeDecodeCycle) {
   LevelDbIndexStateKey key;
 
-  std::vector<std::pair<int32_t, std::string>> ids{
-      {0, "foo/bar"}, {1, "foo/bar2"}, {-1, "foo-bar?baz!quux"}};
+  std::vector<std::pair<std::string, int32_t>> ids{
+      {"foo/bar", 0}, {"foo/bar2", 1}, {"foo-bar?baz!quux", -1}};
   for (auto&& id : ids) {
     auto encoded = LevelDbIndexStateKey::Key(id.first, id.second);
     bool ok = key.Decode(encoded);
     ASSERT_TRUE(ok);
-    ASSERT_EQ(id.first, key.index_id());
-    ASSERT_EQ(id.second, key.user_id());
+    ASSERT_EQ(id.first, key.user_id());
+    ASSERT_EQ(id.second, key.index_id());
   }
 }
 
 TEST(IndexStateKeyTest, Description) {
   AssertExpectedKeyDescription(
-      "[index_state: index_id=99 user_id=foo-bar?baz!quux]",
-      LevelDbIndexStateKey::Key(99, "foo-bar?baz!quux"));
+      "[index_state: user_id=foo-bar?baz!quux index_id=99]",
+      LevelDbIndexStateKey::Key("foo-bar?baz!quux", 99));
 }
 
 TEST(IndexEntryKeyTest, Prefixing) {
@@ -570,6 +588,10 @@ TEST(IndexEntryKeyTest, Prefixing) {
                                 "directional_value_encoded", "document_id_99"),
       table_key));
 
+  ASSERT_TRUE(
+      absl::StartsWith(LevelDbIndexEntryKey::Key(0, "user_id", "", "", ""),
+                       LevelDbIndexEntryKey::KeyPrefix(0)));
+
   ASSERT_FALSE(absl::StartsWith(LevelDbIndexEntryKey::Key(0, "", "", "", ""),
                                 LevelDbIndexEntryKey::Key(1, "", "", "", "")));
 }
diff --git a/Firestore/core/test/unit/testutil/testutil.cc b/Firestore/core/test/unit/testutil/testutil.cc
index b69962b40aa..63d6fabc4d8 100644
--- a/Firestore/core/test/unit/testutil/testutil.cc
+++ b/Firestore/core/test/unit/testutil/testutil.cc
@@ -524,6 +524,17 @@ model::FieldIndex MakeFieldIndex(const std::string& collection_group,
           model::FieldIndex::InitialState()};
 }
 
+model::FieldIndex MakeFieldIndex(const std::string& collection_group,
+                                 int32_t index_id,
+                                 model::IndexState state,
+                                 const std::string& field_1,
+                                 model::Segment::Kind kind_1) {
+  return {index_id,
+          collection_group,
+          {model::Segment{Field(field_1), kind_1}},
+          state};
+}
+
 }  // namespace testutil
 }  // namespace firestore
 }  // namespace firebase
diff --git a/Firestore/core/test/unit/testutil/testutil.h b/Firestore/core/test/unit/testutil/testutil.h
index 09adb32d6db..6cd92ef4071 100644
--- a/Firestore/core/test/unit/testutil/testutil.h
+++ b/Firestore/core/test/unit/testutil/testutil.h
@@ -459,6 +459,11 @@ model::FieldIndex MakeFieldIndex(const std::string& collection_group,
                                  model::Segment::Kind kind_1,
                                  const std::string& field_2,
                                  model::Segment::Kind kind_2);
+model::FieldIndex MakeFieldIndex(const std::string& collection_group,
+                                 int32_t index_id,
+                                 model::IndexState state,
+                                 const std::string& field_1,
+                                 model::Segment::Kind kind_1);
 
 }  // namespace testutil
 }  // namespace firestore