Implement Firestore IndexBackfiller

packages/firestore/src/local/indexeddb_index_manager.ts

@@ -51,7 +51,7 @@ import { FieldPath, ResourcePath } from '../model/path';
 import { TargetIndexMatcher } from '../model/target_index_matcher';
 import { isArray, refValue } from '../model/values';
 import { Value as ProtoValue } from '../protos/firestore_proto_api';
-import {debugAssert, hardAssert} from '../util/assert';
+import { debugAssert, fail, hardAssert } from '../util/assert';
 import { logDebug } from '../util/log';
 import { immediateSuccessor, primitiveComparator } from '../util/misc';
 import { ObjectMap } from '../util/obj_map';
@@ -970,53 +970,22 @@ export class IndexedDbIndexManager implements IndexManager {
     transaction: PersistenceTransaction,
     collectionGroup: string
   ): PersistencePromise<IndexOffset> {
-    return this.getFieldIndexes(transaction, collectionGroup)
-      .next(fieldIndexes => this.getMinOffsetFromFieldIndexes(fieldIndexes));
-  }
-
-  getMinOffsetFromFieldIndexes(fieldIndexes: FieldIndex[]): IndexOffset {
-    hardAssert(
-      fieldIndexes.length !== 0,
-      "Found empty index group when looking for least recent index offset.");
-
-    let minOffset: IndexOffset = fieldIndexes[0].indexState.offset;
-    let maxBatchId: number = minOffset.largestBatchId;
-    for (const fieldIndex of fieldIndexes) {
-      const newOffset: IndexOffset = fieldIndex.indexState.offset;
-      if (indexOffsetComparator(newOffset, minOffset) < 0) {
-        minOffset = newOffset;
-      }
-      if (maxBatchId < newOffset.largestBatchId) {
-        maxBatchId = newOffset.largestBatchId;
-      }
-    }
-    return new IndexOffset(
-      minOffset.readTime,
-      minOffset.documentKey,
-      maxBatchId
+    return this.getFieldIndexes(transaction, collectionGroup).next(
+      getMinOffsetFromFieldIndexes
     );
   }
 
   getMinOffset(
     transaction: PersistenceTransaction,
     target: Target
   ): PersistencePromise<IndexOffset> {
-    let offset: IndexOffset | undefined;
-    return PersistencePromise.forEach(
+    return PersistencePromise.mapArray(
       this.getSubTargets(target),
-      (subTarget: Target) => {
-        return this.getFieldIndex(transaction, subTarget).next(index => {
-          if (!index) {
-            offset = IndexOffset.min();
-          } else if (
-            !offset ||
-            indexOffsetComparator(index.indexState.offset, offset) < 0
-          ) {
-            offset = index.indexState.offset;
-          }
-        });
-      }
-    ).next(() => offset!);
+      (subTarget: Target) =>
+        this.getFieldIndex(transaction, subTarget).next(index =>
+          !index ? fail('Target cannot be served from index') : index
+        )
+    ).next(getMinOffsetFromFieldIndexes);
   }
 }
 
@@ -1062,3 +1031,23 @@ function indexStateStore(
 ): SimpleDbStore<DbIndexStateKey, DbIndexState> {
   return getStore<DbIndexStateKey, DbIndexState>(txn, DbIndexStateStore);
 }
+
+function getMinOffsetFromFieldIndexes(fieldIndexes: FieldIndex[]): IndexOffset {
+  hardAssert(
+    fieldIndexes.length !== 0,
+    'Found empty index group when looking for least recent index offset.'
+  );
+
+  let minOffset: IndexOffset = fieldIndexes[0].indexState.offset;
+  let maxBatchId: number = minOffset.largestBatchId;
+  for (let i = 1; i < fieldIndexes.length; i++) {
+    const newOffset: IndexOffset = fieldIndexes[i].indexState.offset;
+    if (indexOffsetComparator(newOffset, minOffset) < 0) {
+      minOffset = newOffset;
+    }
+    if (maxBatchId < newOffset.largestBatchId) {
+      maxBatchId = newOffset.largestBatchId;
+    }
+  }
+  return new IndexOffset(minOffset.readTime, minOffset.documentKey, maxBatchId);
+}

packages/firestore/src/local/local_documents_view.ts

@@ -373,6 +373,21 @@ export class LocalDocumentsView {
     }
   }
 
+  /**
+   * Given a collection group, returns the next documents that follow the provided offset, along
+   * with an updated batch ID.
+   *
+   * <p>The documents returned by this method are ordered by remote version from the provided
+   * offset. If there are no more remote documents after the provided offset, documents with
+   * mutations in order of batch id from the offset are returned. Since all documents in a batch are
+   * returned together, the total number of documents returned can exceed {@code count}.
+   *
+   * @param transaction
+   * @param collectionGroup The collection group for the documents.
+   * @param offset The offset to index into.
+   * @param count The number of documents to return
+   * @return A LocalWriteResult with the documents that follow the provided offset and the last processed batch id.
+   */
   getNextDocuments(
     transaction: PersistenceTransaction,
     collectionGroup: string,
@@ -391,6 +406,10 @@ export class LocalDocumentsView {
                 count - originalDocs.size
               )
             : PersistencePromise.resolve(newOverlayMap());
+        // The callsite will use the largest batch ID together with the latest read time to create
+        // a new index offset. Since we only process batch IDs if all remote documents have been read,
+        // no overlay will increase the overall read time. This is why we only need to special case
+        // the batch id.
         let largestBatchId = INITIAL_LARGEST_BATCH_ID;
         let modifiedDocs = originalDocs;
         return overlaysPromise.next(overlays => {

packages/firestore/src/local/persistence_promise.ts

@@ -244,6 +244,33 @@ export class PersistencePromise<T> {
     return this.waitFor(promises);
   }
 
+  /**
+   * Concurrently map all array elements through asynchronous function.
+   */
+  static mapArray<T, U>(
+    array: T[],
+    f: (t: T) => PersistencePromise<U>
+  ): PersistencePromise<U[]> {
+    return new PersistencePromise<U[]>((resolve, reject) => {
+      const expectedCount = array.length;
+      const results: U[] = new Array(expectedCount);
+      let resolvedCount = 0;
+      for (let i = 0; i < expectedCount; i++) {
+        const current = i;
+        f(array[current]).next(
+          result => {
+            results[current] = result;
+            ++resolvedCount;
+            if (resolvedCount === expectedCount) {
+              resolve(results);
+            }
+          },
+          err => reject(err)
+        );
+      }
+    });
+  }
+
   /**
    * An alternative to recursive PersistencePromise calls, that avoids
    * potential memory problems from unbounded chains of promises.
@@ -255,11 +282,11 @@ export class PersistencePromise<T> {
     action: () => PersistencePromise<void>
   ): PersistencePromise<void> {
     return new PersistencePromise<void>((resolve, reject) => {
-      const process = () => {
+      const process = (): void => {
         if (condition() === true) {
           action().next(() => {
             process();
-          }, reject)
+          }, reject);
         } else {
           resolve();
         }