Use DocumentKey as tie breaker in index backfill

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

@@ -21,6 +21,7 @@
 import com.google.firebase.firestore.core.Query;
 import com.google.firebase.firestore.model.Document;
 import com.google.firebase.firestore.model.DocumentKey;
+import com.google.firebase.firestore.model.FieldIndex.IndexOffset;
 import com.google.firebase.firestore.model.SnapshotVersion;
 import com.google.firebase.firestore.util.Logger;
 import java.util.Collections;
@@ -98,7 +99,8 @@ && needsRefill(
     // Retrieve all results for documents that were updated since the last limbo-document free
     // remote snapshot.
     ImmutableSortedMap<DocumentKey, Document> updatedResults =
-        localDocumentsView.getDocumentsMatchingQuery(query, lastLimboFreeSnapshotVersion);
+        localDocumentsView.getDocumentsMatchingQuery(
+            query, IndexOffset.create(lastLimboFreeSnapshotVersion));
 
     // We merge `previousResults` into `updateResults`, since `updateResults` is already a
     // ImmutableSortedMap. If a document is contained in both lists, then its contents are the same.
@@ -168,6 +170,6 @@ private ImmutableSortedMap<DocumentKey, Document> executeFullCollectionScan(Quer
     if (Logger.isDebugEnabled()) {
       Logger.debug(LOG_TAG, "Using full collection scan to execute query: %s", query.toString());
     }
-    return localDocumentsView.getDocumentsMatchingQuery(query, SnapshotVersion.NONE);
+    return localDocumentsView.getDocumentsMatchingQuery(query, IndexOffset.NONE);
   }
 }

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

@@ -23,8 +23,8 @@
 import com.google.firebase.firestore.model.Document;
 import com.google.firebase.firestore.model.DocumentKey;
 import com.google.firebase.firestore.model.FieldIndex;
+import com.google.firebase.firestore.model.FieldIndex.IndexOffset;
 import com.google.firebase.firestore.model.ResourcePath;
-import com.google.firebase.firestore.model.SnapshotVersion;
 import com.google.firebase.firestore.util.AsyncQueue;
 import com.google.firebase.firestore.util.Logger;
 import java.util.ArrayList;
@@ -140,42 +140,53 @@ private int writeEntriesForCollectionGroup(
       LocalDocumentsView localDocumentsView, String collectionGroup, int entriesRemainingUnderCap) {
     Query query = new Query(ResourcePath.EMPTY, collectionGroup);
 
-    // Use the earliest readTime of all field indexes as the base readtime.
-    SnapshotVersion earliestReadTime =
-        getEarliestReadTime(indexManager.getFieldIndexes(collectionGroup));
+    // Use the earliest offset of all field indexes to query the local cache.
+    IndexOffset existingOffset = getExistingOffset(indexManager.getFieldIndexes(collectionGroup));
 
     // TODO(indexing): Use limit queries to only fetch the required number of entries.
     // TODO(indexing): Support mutation batch Ids when sorting and writing indexes.
     ImmutableSortedMap<DocumentKey, Document> documents =
-        localDocumentsView.getDocumentsMatchingQuery(query, earliestReadTime);
+        localDocumentsView.getDocumentsMatchingQuery(query, existingOffset);
 
     List<Document> oldestDocuments = getOldestDocuments(documents, entriesRemainingUnderCap);
     indexManager.updateIndexEntries(oldestDocuments);
 
-    SnapshotVersion latestReadTime = getPostUpdateReadTime(oldestDocuments, earliestReadTime);
-    indexManager.updateCollectionGroup(collectionGroup, latestReadTime);
+    IndexOffset newOffset = getNewOffset(oldestDocuments, existingOffset);
+    indexManager.updateCollectionGroup(collectionGroup, newOffset);
     return oldestDocuments.size();
   }
 
+  /** Returns the lowest offset for the provided index group. */
+  private IndexOffset getExistingOffset(Collection<FieldIndex> fieldIndexes) {
+    IndexOffset lowestOffset = null;
+    for (FieldIndex fieldIndex : fieldIndexes) {
+      if (lowestOffset == null
+          || fieldIndex.getIndexState().getOffset().compareTo(lowestOffset) < 0) {
+        lowestOffset = fieldIndex.getIndexState().getOffset();
+      }
+    }
+    return lowestOffset == null ? IndexOffset.NONE : lowestOffset;
+  }
+
   /**
-   * Returns the new read time for the index.
+   * Returns the offset for the index based on the newly indexed documents.
    *
-   * @param documents a list of documents sorted by read time (ascending)
-   * @param currentReadTime the current read time of the index
+   * @param documents a list of documents sorted by read time and key (ascending)
+   * @param currentOffset the current offset of the index group
    */
-  private SnapshotVersion getPostUpdateReadTime(
-      List<Document> documents, SnapshotVersion currentReadTime) {
-    SnapshotVersion latestReadTime =
+  private IndexOffset getNewOffset(List<Document> documents, IndexOffset currentOffset) {
+    IndexOffset latestOffset =
         documents.isEmpty()
-            ? remoteDocumentCache.getLatestReadTime()
-            : documents.get(documents.size() - 1).getReadTime();
+            ? IndexOffset.create(remoteDocumentCache.getLatestReadTime())
+            : IndexOffset.create(
+                documents.get(documents.size() - 1).getReadTime(),
+                documents.get(documents.size() - 1).getKey());
     // Make sure the index does not go back in time
-    latestReadTime =
-        latestReadTime.compareTo(currentReadTime) > 0 ? latestReadTime : currentReadTime;
-    return latestReadTime;
+    latestOffset = latestOffset.compareTo(currentOffset) > 0 ? latestOffset : currentOffset;
+    return latestOffset;
   }
 
-  /** Returns up to {@code count} documents sorted by read time. */
+  /** Returns up to {@code count} documents sorted by read time and key. */
   private List<Document> getOldestDocuments(
       ImmutableSortedMap<DocumentKey, Document> documents, int count) {
     List<Document> oldestDocuments = new ArrayList<>();
@@ -184,27 +195,12 @@ private List<Document> getOldestDocuments(
     }
     Collections.sort(
         oldestDocuments,
-        (l, r) -> {
-          int cmp = l.getReadTime().compareTo(r.getReadTime());
-          if (cmp != 0) return cmp;
-          return l.getKey().compareTo(r.getKey());
-        });
+        (l, r) ->
+            IndexOffset.create(l.getReadTime(), l.getKey())
+                .compareTo(IndexOffset.create(r.getReadTime(), r.getKey())));
     return oldestDocuments.subList(0, Math.min(count, oldestDocuments.size()));
   }
 
-  private SnapshotVersion getEarliestReadTime(Collection<FieldIndex> fieldIndexes) {
-    SnapshotVersion lowestVersion = null;
-    for (FieldIndex fieldIndex : fieldIndexes) {
-      lowestVersion =
-          lowestVersion == null
-              ? fieldIndex.getIndexState().getReadTime()
-              : fieldIndex.getIndexState().getReadTime().compareTo(lowestVersion) < 0
-                  ? fieldIndex.getIndexState().getReadTime()
-                  : lowestVersion;
-    }
-    return lowestVersion;
-  }
-
   @VisibleForTesting
   void setMaxDocumentsToProcess(int newMax) {
     maxDocumentsToProcess = newMax;

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

@@ -20,7 +20,6 @@
 import com.google.firebase.firestore.model.DocumentKey;
 import com.google.firebase.firestore.model.FieldIndex;
 import com.google.firebase.firestore.model.ResourcePath;
-import com.google.firebase.firestore.model.SnapshotVersion;
 import java.util.Collection;
 import java.util.List;
 import java.util.Set;
@@ -91,11 +90,11 @@ public interface IndexManager {
   /**
    * Sets the collection group's latest read time.
    *
-   * <p>This method updates the read time for all field indices for the collection group and
+   * <p>This method updates the index offset for all field indices for the collection group and
    * increments their sequence number. Subsequent calls to {@link #getNextCollectionGroupToUpdate()}
    * will return a different collection group (unless only one collection group is configured).
    */
-  void updateCollectionGroup(String collectionGroup, SnapshotVersion readTime);
+  void updateCollectionGroup(String collectionGroup, FieldIndex.IndexOffset offset);
 
   /** Updates the index entries for the provided documents. */
   void updateIndexEntries(Collection<Document> documents);

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

@@ -58,7 +58,7 @@ public ImmutableSortedMap<DocumentKey, Document> getDocumentsMatchingQuery(
     hardAssert(localDocuments != null, "setLocalDocumentsView() not called");
 
     return query.isDocumentQuery()
-        ? localDocuments.getDocumentsMatchingQuery(query, SnapshotVersion.NONE)
+        ? localDocuments.getDocumentsMatchingQuery(query, FieldIndex.IndexOffset.NONE)
         : performCollectionQuery(query);
   }
 
@@ -84,7 +84,7 @@ private ImmutableSortedMap<DocumentKey, Document> performCollectionQuery(Query q
       ImmutableSortedMap<DocumentKey, Document> indexedDocuments =
           localDocuments.getDocuments(keys);
       ImmutableSortedMap<DocumentKey, Document> additionalDocuments =
-          localDocuments.getDocumentsMatchingQuery(query, fieldIndex.getIndexState().getReadTime());
+          localDocuments.getDocumentsMatchingQuery(query, fieldIndex.getIndexState().getOffset());
       for (Map.Entry<DocumentKey, Document> entry : additionalDocuments) {
         indexedDocuments = indexedDocuments.insert(entry.getKey(), entry.getValue());
       }
@@ -98,6 +98,6 @@ private ImmutableSortedMap<DocumentKey, Document> executeFullCollectionScan(Quer
     if (Logger.isDebugEnabled()) {
       Logger.debug(LOG_TAG, "Using full collection scan to execute query: %s", query.toString());
     }
-    return localDocuments.getDocumentsMatchingQuery(query, SnapshotVersion.NONE);
+    return localDocuments.getDocumentsMatchingQuery(query, FieldIndex.IndexOffset.NONE);
   }
 }

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

@@ -23,9 +23,9 @@
 import com.google.firebase.firestore.core.Query;
 import com.google.firebase.firestore.model.Document;
 import com.google.firebase.firestore.model.DocumentKey;
+import com.google.firebase.firestore.model.FieldIndex.IndexOffset;
 import com.google.firebase.firestore.model.MutableDocument;
 import com.google.firebase.firestore.model.ResourcePath;
-import com.google.firebase.firestore.model.SnapshotVersion;
 import com.google.firebase.firestore.model.mutation.FieldMask;
 import com.google.firebase.firestore.model.mutation.Mutation;
 import com.google.firebase.firestore.model.mutation.MutationBatch;
@@ -246,18 +246,17 @@ void recalculateAndSaveOverlays(Set<DocumentKey> documentKeys) {
    * Performs a query against the local view of all documents.
    *
    * @param query The query to match documents against.
-   * @param sinceReadTime If not set to SnapshotVersion.MIN, return only documents that have been
-   *     read since this snapshot version (exclusive).
+   * @param offset Read time and key to start scanning by (exclusive).
    */
   ImmutableSortedMap<DocumentKey, Document> getDocumentsMatchingQuery(
-      Query query, SnapshotVersion sinceReadTime) {
+      Query query, IndexOffset offset) {
     ResourcePath path = query.getPath();
     if (query.isDocumentQuery()) {
       return getDocumentsMatchingDocumentQuery(path);
     } else if (query.isCollectionGroupQuery()) {
-      return getDocumentsMatchingCollectionGroupQuery(query, sinceReadTime);
+      return getDocumentsMatchingCollectionGroupQuery(query, offset);
     } else {
-      return getDocumentsMatchingCollectionQuery(query, sinceReadTime);
+      return getDocumentsMatchingCollectionQuery(query, offset);
     }
   }
 
@@ -274,7 +273,7 @@ private ImmutableSortedMap<DocumentKey, Document> getDocumentsMatchingDocumentQu
   }
 
   private ImmutableSortedMap<DocumentKey, Document> getDocumentsMatchingCollectionGroupQuery(
-      Query query, SnapshotVersion sinceReadTime) {
+      Query query, IndexOffset offset) {
     hardAssert(
         query.getPath().isEmpty(),
         "Currently we only support collection group queries at the root.");
@@ -287,7 +286,7 @@ private ImmutableSortedMap<DocumentKey, Document> getDocumentsMatchingCollection
     for (ResourcePath parent : parents) {
       Query collectionQuery = query.asCollectionQueryAtPath(parent.append(collectionId));
       ImmutableSortedMap<DocumentKey, Document> collectionResults =
-          getDocumentsMatchingCollectionQuery(collectionQuery, sinceReadTime);
+          getDocumentsMatchingCollectionQuery(collectionQuery, offset);
       for (Map.Entry<DocumentKey, Document> docEntry : collectionResults) {
         results = results.insert(docEntry.getKey(), docEntry.getValue());
       }
@@ -296,11 +295,11 @@ private ImmutableSortedMap<DocumentKey, Document> getDocumentsMatchingCollection
   }
 
   private ImmutableSortedMap<DocumentKey, Document> getDocumentsMatchingCollectionQuery(
-      Query query, SnapshotVersion sinceReadTime) {
+      Query query, IndexOffset offset) {
     if (Persistence.OVERLAY_SUPPORT_ENABLED) {
       // TODO(Overlay): Remove the assert and just return `fromOverlay`.
       ImmutableSortedMap<DocumentKey, Document> fromOverlay =
-          getDocumentsMatchingCollectionQueryFromOverlayCache(query, sinceReadTime);
+          getDocumentsMatchingCollectionQueryFromOverlayCache(query, offset);
       // TODO(Overlay): Delete below before merging. The code passes, but there are tests
       // looking at how many documents read from remote document, and this would double
       // the count.
@@ -313,16 +312,15 @@ private ImmutableSortedMap<DocumentKey, Document> getDocumentsMatchingCollection
        */
       return fromOverlay;
     } else {
-      return getDocumentsMatchingCollectionQueryFromMutationQueue(query, sinceReadTime);
+      return getDocumentsMatchingCollectionQueryFromMutationQueue(query, offset);
     }
   }
 
   /** Queries the remote documents and overlays by doing a full collection scan. */
   private ImmutableSortedMap<DocumentKey, Document>
-      getDocumentsMatchingCollectionQueryFromOverlayCache(
-          Query query, SnapshotVersion sinceReadTime) {
+      getDocumentsMatchingCollectionQueryFromOverlayCache(Query query, IndexOffset offset) {
     ImmutableSortedMap<DocumentKey, MutableDocument> remoteDocuments =
-        remoteDocumentCache.getAllDocumentsMatchingQuery(query, sinceReadTime);
+        remoteDocumentCache.getAllDocumentsMatchingQuery(query, offset);
     Map<DocumentKey, Mutation> overlays = documentOverlayCache.getOverlays(query.getPath(), -1);
 
     // As documents might match the query because of their overlay we need to include all documents
@@ -356,10 +354,9 @@ private ImmutableSortedMap<DocumentKey, Document> getDocumentsMatchingCollection
 
   /** Queries the remote documents and mutation queue, by doing a full collection scan. */
   private ImmutableSortedMap<DocumentKey, Document>
-      getDocumentsMatchingCollectionQueryFromMutationQueue(
-          Query query, SnapshotVersion sinceReadTime) {
+      getDocumentsMatchingCollectionQueryFromMutationQueue(Query query, IndexOffset offset) {
     ImmutableSortedMap<DocumentKey, MutableDocument> remoteDocuments =
-        remoteDocumentCache.getAllDocumentsMatchingQuery(query, sinceReadTime);
+        remoteDocumentCache.getAllDocumentsMatchingQuery(query, offset);
 
     // TODO(indexing): We should plumb sinceReadTime through to the mutation queue
     List<MutationBatch> matchingBatches = mutationQueue.getAllMutationBatchesAffectingQuery(query);

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

@@ -21,7 +21,6 @@
 import com.google.firebase.firestore.model.DocumentKey;
 import com.google.firebase.firestore.model.FieldIndex;
 import com.google.firebase.firestore.model.ResourcePath;
-import com.google.firebase.firestore.model.SnapshotVersion;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -82,7 +81,7 @@ public String getNextCollectionGroupToUpdate() {
   }
 
   @Override
-  public void updateCollectionGroup(String collectionGroup, SnapshotVersion readTime) {
+  public void updateCollectionGroup(String collectionGroup, FieldIndex.IndexOffset offset) {
     // Field indices are not supported with memory persistence.
   }
 

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

@@ -22,6 +22,7 @@
 import com.google.firebase.database.collection.ImmutableSortedMap;
 import com.google.firebase.firestore.core.Query;
 import com.google.firebase.firestore.model.DocumentKey;
+import com.google.firebase.firestore.model.FieldIndex.IndexOffset;
 import com.google.firebase.firestore.model.MutableDocument;
 import com.google.firebase.firestore.model.ResourcePath;
 import com.google.firebase.firestore.model.SnapshotVersion;
@@ -83,7 +84,7 @@ public Map<DocumentKey, MutableDocument> getAll(Iterable<DocumentKey> keys) {
 
   @Override
   public ImmutableSortedMap<DocumentKey, MutableDocument> getAllDocumentsMatchingQuery(
-      Query query, SnapshotVersion sinceReadTime) {
+      Query query, IndexOffset offset) {
     hardAssert(
         !query.isCollectionGroupQuery(),
         "CollectionGroup queries should be handled in LocalDocumentsView");
@@ -110,7 +111,7 @@ public ImmutableSortedMap<DocumentKey, MutableDocument> getAllDocumentsMatchingQ
       }
 
       SnapshotVersion readTime = entry.getValue().second;
-      if (readTime.compareTo(sinceReadTime) <= 0) {
+      if (IndexOffset.create(readTime, doc.getKey()).compareTo(offset) <= 0) {
         continue;
       }
 

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

@@ -17,6 +17,7 @@
 import com.google.firebase.database.collection.ImmutableSortedMap;
 import com.google.firebase.firestore.core.Query;
 import com.google.firebase.firestore.model.DocumentKey;
+import com.google.firebase.firestore.model.FieldIndex;
 import com.google.firebase.firestore.model.MutableDocument;
 import com.google.firebase.firestore.model.SnapshotVersion;
 import java.util.Map;
@@ -73,12 +74,11 @@ interface RemoteDocumentCache {
    * <p>Cached entries for non-existing documents have no bearing on query results.
    *
    * @param query The query to match documents against.
-   * @param sinceReadTime If not set to SnapshotVersion.MIN, return only documents that have been
-   *     read since this snapshot version (exclusive).
+   * @param offset The read time and document key to start scanning at (exclusive).
    * @return The set of matching documents.
    */
   ImmutableSortedMap<DocumentKey, MutableDocument> getAllDocumentsMatchingQuery(
-      Query query, SnapshotVersion sinceReadTime);
+      Query query, FieldIndex.IndexOffset offset);
 
   /** Returns the latest read time of any document in the cache. */
   SnapshotVersion getLatestReadTime();