Add largest batch ID

firebase-firestore/src/main/java/com/google/firebase/firestore/local/LocalDocumentsView.java

@@ -260,7 +260,8 @@ private ImmutableSortedMap<DocumentKey, Document> getDocumentsMatchingCollection
       Query query, IndexOffset offset) {
     Map<DocumentKey, MutableDocument> remoteDocuments =
         remoteDocumentCache.getAll(query.getPath(), offset);
-    Map<DocumentKey, Overlay> overlays = documentOverlayCache.getOverlays(query.getPath(), -1);
+    Map<DocumentKey, Overlay> overlays =
+        documentOverlayCache.getOverlays(query.getPath(), offset.getLargestBatchId());
 
     // As documents might match the query because of their overlay we need to include documents
     // for all overlays in the initial document set.

firebase-firestore/src/main/java/com/google/firebase/firestore/local/SQLiteIndexManager.java

@@ -106,8 +106,8 @@ public void start() {
     // Fetch all index states if persisted for the user. These states contain per user information
     // on how up to date the index is.
     db.query(
-            "SELECT index_id, sequence_number, read_time_seconds, read_time_nanos, document_key "
-                + "FROM index_state WHERE uid = ?")
+            "SELECT index_id, sequence_number, read_time_seconds, read_time_nanos, document_key, "
+                + "largest_batch_id FROM index_state WHERE uid = ?")
         .binding(uid)
         .forEach(
             row -> {
@@ -117,8 +117,11 @@ public void start() {
                   new SnapshotVersion(new Timestamp(row.getLong(2), row.getInt(3)));
               DocumentKey documentKey =
                   DocumentKey.fromPath(EncodedPath.decodeResourcePath(row.getString(4)));
+              int largestBatchId = row.getInt(5);
               indexStates.put(
-                  indexId, FieldIndex.IndexState.create(sequenceNumber, readTime, documentKey));
+                  indexId,
+                  FieldIndex.IndexState.create(
+                      sequenceNumber, readTime, documentKey, largestBatchId));
             });
 
     // Fetch all indices and combine with user's index state if available.
@@ -660,13 +663,15 @@ public void updateCollectionGroup(String collectionGroup, FieldIndex.IndexOffset
               FieldIndex.IndexState.create(memoizedMaxSequenceNumber, offset));
       db.execute(
           "REPLACE INTO index_state (index_id, uid,  sequence_number, "
-              + "read_time_seconds, read_time_nanos, document_key) VALUES(?, ?, ?, ?, ?, ?)",
+              + "read_time_seconds, read_time_nanos, document_key, largest_batch_id) "
+              + "VALUES(?, ?, ?, ?, ?, ?, ?)",
           fieldIndex.getIndexId(),
           uid,
           memoizedMaxSequenceNumber,
           offset.getReadTime().getTimestamp().getSeconds(),
           offset.getReadTime().getTimestamp().getNanoseconds(),
-          EncodedPath.encode(offset.getDocumentKey().getPath()));
+          EncodedPath.encode(offset.getDocumentKey().getPath()),
+          offset.getLargestBatchId());
       memoizeIndex(updatedIndex);
     }
   }

firebase-firestore/src/main/java/com/google/firebase/firestore/local/SQLiteSchema.java

@@ -398,6 +398,7 @@ private void createFieldIndex() {
                   + "read_time_seconds INTEGER, " // Read time of last processed document
                   + "read_time_nanos INTEGER, "
                   + "document_key TEXT, " // Key of the last processed document
+                  + "largest_batch_id INTEGER, " // Largest mutation batch id that was processed
                   + "PRIMARY KEY (index_id, uid))");
 
           // The index entry table stores the encoded entries for all fields.

firebase-firestore/src/main/java/com/google/firebase/firestore/model/FieldIndex.java

@@ -39,12 +39,15 @@ public abstract class FieldIndex {
   /** An ID for an index that has not yet been added to persistence. */
   public static final int UNKNOWN_ID = -1;
 
+  /** The initial mutation batch id for each index. Gets updated during index backfill. */
+  public static final int INITIAL_LARGEST_BATCH_ID = -1;
+
   /** The initial sequence number for each index. Gets updated during index backfill. */
   public static final int INITIAL_SEQUENCE_NUMBER = 0;
 
   /** The state of an index that has not yet been backfilled. */
   public static IndexState INITIAL_STATE =
-      IndexState.create(INITIAL_SEQUENCE_NUMBER, SnapshotVersion.NONE, DocumentKey.empty());
+      IndexState.create(INITIAL_SEQUENCE_NUMBER, IndexOffset.NONE);
 
   /** Compares indexes by collection group and segments. Ignores update time and index ID. */
   public static final Comparator<FieldIndex> SEMANTIC_COMPARATOR =
@@ -100,8 +103,11 @@ public static IndexState create(long sequenceNumber, IndexOffset offset) {
     }
 
     public static IndexState create(
-        long sequenceNumber, SnapshotVersion readTime, DocumentKey documentKey) {
-      return create(sequenceNumber, IndexOffset.create(readTime, documentKey));
+        long sequenceNumber,
+        SnapshotVersion readTime,
+        DocumentKey documentKey,
+        int largestBatchId) {
+      return create(sequenceNumber, IndexOffset.create(readTime, documentKey, largestBatchId));
     }
 
     /**
@@ -116,17 +122,20 @@ public static IndexState create(
   /** Stores the latest read time and document that were processed for an index. */
   @AutoValue
   public abstract static class IndexOffset implements Comparable<IndexOffset> {
+    public static final IndexOffset NONE =
+        create(SnapshotVersion.NONE, DocumentKey.empty(), INITIAL_LARGEST_BATCH_ID);
+
     public static final Comparator<MutableDocument> DOCUMENT_COMPARATOR =
         (l, r) -> IndexOffset.fromDocument(l).compareTo(IndexOffset.fromDocument(r));
 
-    public static final IndexOffset NONE = create(SnapshotVersion.NONE, DocumentKey.empty());
-
     /**
      * Creates an offset that matches all documents with a read time higher than {@code readTime} or
-     * with a key higher than {@code documentKey} for equal read times.
+     * with a key higher than {@code documentKey} for equal read times. The largest batch ID is used
+     * as a final tie breaker.
      */
-    public static IndexOffset create(SnapshotVersion readTime, DocumentKey documentKey) {
-      return new AutoValue_FieldIndex_IndexOffset(readTime, documentKey);
+    public static IndexOffset create(
+        SnapshotVersion readTime, DocumentKey key, int largestBatchId) {
+      return new AutoValue_FieldIndex_IndexOffset(readTime, key, largestBatchId);
     }
 
     /**
@@ -145,12 +154,12 @@ public static IndexOffset create(SnapshotVersion readTime) {
               successorNanos == 1e9
                   ? new Timestamp(successorSeconds + 1, 0)
                   : new Timestamp(successorSeconds, successorNanos));
-      return new AutoValue_FieldIndex_IndexOffset(successor, DocumentKey.empty());
+      return create(successor, DocumentKey.empty(), INITIAL_LARGEST_BATCH_ID);
     }
 
     /** Creates a new offset based on the provided document. */
     public static IndexOffset fromDocument(Document document) {
-      return new AutoValue_FieldIndex_IndexOffset(document.getReadTime(), document.getKey());
+      return create(document.getReadTime(), document.getKey(), INITIAL_LARGEST_BATCH_ID);
     }
 
     /**
@@ -164,10 +173,17 @@ public static IndexOffset fromDocument(Document document) {
      */
     public abstract DocumentKey getDocumentKey();
 
+    /*
+     * Returns the largest mutation batch id that's been processed by Firestore.
+     */
+    public abstract int getLargestBatchId();
+
     public int compareTo(IndexOffset other) {
       int cmp = getReadTime().compareTo(other.getReadTime());
       if (cmp != 0) return cmp;
-      return getDocumentKey().compareTo(other.getDocumentKey());
+      cmp = getDocumentKey().compareTo(other.getDocumentKey());
+      if (cmp != 0) return cmp;
+      return Integer.compare(getLargestBatchId(), other.getLargestBatchId());
     }
   }
 

firebase-firestore/src/test/java/com/google/firebase/firestore/local/IndexBackfillerTest.java

@@ -289,7 +289,7 @@ private void addFieldIndex(String collectionGroup, String fieldName, SnapshotVer
         fieldIndex(
             collectionGroup,
             FieldIndex.UNKNOWN_ID,
-            FieldIndex.IndexState.create(0, version, DocumentKey.empty()),
+            FieldIndex.IndexState.create(0, version, DocumentKey.empty(), -1),
             fieldName,
             FieldIndex.Segment.Kind.ASCENDING);
     indexManager.addFieldIndex(fieldIndex);
@@ -300,7 +300,7 @@ private void addFieldIndex(String collectionGroup, String fieldName, long sequen
         fieldIndex(
             collectionGroup,
             FieldIndex.UNKNOWN_ID,
-            FieldIndex.IndexState.create(sequenceNumber, SnapshotVersion.NONE, DocumentKey.empty()),
+            FieldIndex.IndexState.create(sequenceNumber, IndexOffset.NONE),
             fieldName,
             FieldIndex.Segment.Kind.ASCENDING);
     indexManager.addFieldIndex(fieldIndex);

firebase-firestore/src/test/java/com/google/firebase/firestore/local/RemoteDocumentCacheTestCase.java

@@ -232,7 +232,7 @@ public void testGetAllFromSinceReadTimeAndDocumentKey() {
 
     ResourcePath collection = path("b");
     Map<DocumentKey, MutableDocument> results =
-        remoteDocumentCache.getAll(collection, IndexOffset.create(version(11), key("b/b")));
+        remoteDocumentCache.getAll(collection, IndexOffset.create(version(11), key("b/b"), -1));
     assertThat(results.values()).containsExactly(doc("b/c", 3, DOC_DATA), doc("b/d", 4, DOC_DATA));
   }
 

firebase-firestore/src/test/java/com/google/firebase/firestore/local/SQLiteIndexManagerTest.java

@@ -582,19 +582,18 @@ public void testAdvancedQueries() {
   }
 
   @Test
-  public void testUpdateTime() {
-    indexManager.addFieldIndex(
-        fieldIndex(
-            "coll1",
-            1,
-            IndexState.create(-1, version(20), DocumentKey.empty()),
-            "value",
-            Kind.ASCENDING));
+  public void testPersistsIndexOffset() {
+    indexManager.addFieldIndex(fieldIndex("coll1", "value", Kind.ASCENDING));
+    IndexOffset offset = IndexOffset.create(version(20), key("coll/doc"), 42);
+    indexManager.updateCollectionGroup("coll1", offset);
+
+    indexManager = persistence.getIndexManager(User.UNAUTHENTICATED);
+    indexManager.start();
 
     Collection<FieldIndex> indexes = indexManager.getFieldIndexes("coll1");
     assertEquals(indexes.size(), 1);
     FieldIndex index = indexes.iterator().next();
-    assertEquals(index.getIndexState().getOffset().getReadTime(), version(20));
+    assertEquals(offset, index.getIndexState().getOffset());
   }
 
   @Test
@@ -638,19 +637,9 @@ public void testGetFieldIndexes() {
   @Test
   public void testDeleteFieldIndexRemovesEntryFromCollectionGroup() {
     indexManager.addFieldIndex(
-        fieldIndex(
-            "coll1",
-            1,
-            IndexState.create(1, version(30), DocumentKey.empty()),
-            "value",
-            Kind.ASCENDING));
+        fieldIndex("coll1", 1, IndexState.create(1, IndexOffset.NONE), "value", Kind.ASCENDING));
     indexManager.addFieldIndex(
-        fieldIndex(
-            "coll2",
-            2,
-            IndexState.create(2, version(0), DocumentKey.empty()),
-            "value",
-            Kind.CONTAINS));
+        fieldIndex("coll2", 2, IndexState.create(2, IndexOffset.NONE), "value", Kind.CONTAINS));
     String collectionGroup = indexManager.getNextCollectionGroupToUpdate();
     assertEquals("coll1", collectionGroup);
 

firebase-firestore/src/test/java/com/google/firebase/firestore/model/FieldIndexTest.java

@@ -54,7 +54,7 @@ public void comparatorIgnoresIndexState() {
     FieldIndex indexOriginal = fieldIndex("collA", 1, FieldIndex.INITIAL_STATE);
     FieldIndex indexSame = fieldIndex("collA", 1, FieldIndex.INITIAL_STATE);
     FieldIndex indexDifferent =
-        fieldIndex("collA", 1, IndexState.create(1, version(2), DocumentKey.empty()));
+        fieldIndex("collA", 1, IndexState.create(1, version(2), DocumentKey.empty(), -1));
     assertEquals(0, SEMANTIC_COMPARATOR.compare(indexOriginal, indexSame));
     assertEquals(0, SEMANTIC_COMPARATOR.compare(indexOriginal, indexDifferent));
   }
@@ -94,10 +94,10 @@ public void comparatorIncludesSegmentsLength() {
 
   @Test
   public void indexOffsetComparator() {
-    IndexOffset docAOffset = IndexOffset.create(version(1), key("foo/a"));
-    IndexOffset docBOffset = IndexOffset.create(version(1), key("foo/b"));
+    IndexOffset docAOffset = IndexOffset.create(version(1), key("foo/a"), -1);
+    IndexOffset docBOffset = IndexOffset.create(version(1), key("foo/b"), -1);
     IndexOffset version1Offset = IndexOffset.create(version(1));
-    IndexOffset docCOffset = IndexOffset.create(version(2), key("foo/c"));
+    IndexOffset docCOffset = IndexOffset.create(version(2), key("foo/c"), -1);
     IndexOffset version2Offset = IndexOffset.create(version(2));
 
     assertEquals(-1, docAOffset.compareTo(docBOffset));
@@ -110,7 +110,7 @@ public void indexOffsetComparator() {
   @Test
   public void indexOffsetAdvancesSeconds() {
     IndexOffset actualSuccessor = IndexOffset.create(version(1, (int) 1e9 - 1));
-    IndexOffset expectedSuccessor = IndexOffset.create(version(2, 0), DocumentKey.empty());
+    IndexOffset expectedSuccessor = IndexOffset.create(version(2, 0), DocumentKey.empty(), -1);
     assertEquals(expectedSuccessor, actualSuccessor);
   }
 }