Track readTime in the RemoteDocument store

Author: schmidt-sebastian

packages/firestore/src/local/indexeddb_persistence.ts

@@ -70,6 +70,7 @@ import { QueryData } from './query_data';
 import { ReferenceSet } from './reference_set';
 import { ClientId } from './shared_client_state';
 import { SimpleDb, SimpleDbStore, SimpleDbTransaction } from './simple_db';
+import { SnapshotVersion } from '../core/snapshot_version';
 
 const LOG_TAG = 'IndexedDbPersistence';
 
@@ -1263,7 +1264,7 @@ export class IndexedDbLruDelegate implements ReferenceDelegate, LruDelegate {
 
     return iteration
       .next(() => PersistencePromise.waitFor(promises))
-      .next(() => changeBuffer.apply(txn))
+      .next(() => changeBuffer.apply(txn, SnapshotVersion.forDeletedDoc()))
       .next(() => documentCount);
   }
 

packages/firestore/src/local/indexeddb_remote_document_cache.ts

@@ -462,7 +462,8 @@ export class IndexedDbRemoteDocumentCache implements RemoteDocumentCache {
     }
 
     protected applyChanges(
-      transaction: PersistenceTransaction
+      transaction: PersistenceTransaction,
+      readTime: SnapshotVersion
     ): PersistencePromise<void> {
       const promises: Array<PersistencePromise<void>> = [];
 
@@ -477,7 +478,8 @@ export class IndexedDbRemoteDocumentCache implements RemoteDocumentCache {
         );
         if (maybeDocument) {
           const doc = this.documentCache.serializer.toDbRemoteDocument(
-            maybeDocument
+            maybeDocument,
+            readTime
           );
           const size = dbDocumentSize(doc);
           sizeDelta += size - previousSize!;

packages/firestore/src/local/indexeddb_schema.ts

@@ -45,8 +45,10 @@ import { SimpleDbSchemaConverter, SimpleDbTransaction } from './simple_db';
  * 6. Create document global for tracking document cache size.
  * 7. Ensure every cached document has a sentinel row with a sequence number.
  * 8. Add collection-parent index for Collection Group queries.
+ * 9. Change RemoteDocumentChanges store to be keyed by readTime rather than
+ *    an auto-incrementing ID. This is required for Index-Free queries.
  */
-export const SCHEMA_VERSION = 8;
+export const SCHEMA_VERSION = 9;
 
 /** Performs database creation and schema upgrades. */
 export class SchemaConverter implements SimpleDbSchemaConverter {
@@ -61,7 +63,7 @@ export class SchemaConverter implements SimpleDbSchemaConverter {
    */
   createOrUpgrade(
     db: IDBDatabase,
-    txn: SimpleDbTransaction,
+    txn: IDBTransaction,
     fromVersion: number,
     toVersion: number
   ): PersistencePromise<void> {
@@ -72,6 +74,8 @@ export class SchemaConverter implements SimpleDbSchemaConverter {
       `Unexpected schema upgrade from v${fromVersion} to v{toVersion}.`
     );
 
+    const simpleDbTransaction = new SimpleDbTransaction(txn);
+
     if (fromVersion < 1 && toVersion >= 1) {
       createPrimaryClientStore(db);
       createMutationQueue(db);
@@ -90,7 +94,7 @@ export class SchemaConverter implements SimpleDbSchemaConverter {
         dropQueryCache(db);
         createQueryCache(db);
       }
-      p = p.next(() => writeEmptyTargetGlobalEntry(txn));
+      p = p.next(() => writeEmptyTargetGlobalEntry(simpleDbTransaction));
     }
 
     if (fromVersion < 4 && toVersion >= 4) {
@@ -101,7 +105,9 @@ export class SchemaConverter implements SimpleDbSchemaConverter {
         // and write them back out. We preserve the existing batch IDs to guarantee
         // consistency with other object stores. Any further mutation batch IDs will
         // be auto-generated.
-        p = p.next(() => upgradeMutationBatchSchemaAndMigrateData(db, txn));
+        p = p.next(() =>
+          upgradeMutationBatchSchemaAndMigrateData(db, simpleDbTransaction)
+        );
       }
 
       p = p.next(() => {
@@ -111,24 +117,31 @@ export class SchemaConverter implements SimpleDbSchemaConverter {
     }
 
     if (fromVersion < 5 && toVersion >= 5) {
-      p = p.next(() => this.removeAcknowledgedMutations(txn));
+      p = p.next(() => this.removeAcknowledgedMutations(simpleDbTransaction));
     }
 
     if (fromVersion < 6 && toVersion >= 6) {
       p = p.next(() => {
         createDocumentGlobalStore(db);
-        return this.addDocumentGlobal(txn);
+        return this.addDocumentGlobal(simpleDbTransaction);
       });
     }
 
     if (fromVersion < 7 && toVersion >= 7) {
-      p = p.next(() => this.ensureSequenceNumbers(txn));
+      p = p.next(() => this.ensureSequenceNumbers(simpleDbTransaction));
     }
 
     if (fromVersion < 8 && toVersion >= 8) {
-      p = p.next(() => this.createCollectionParentIndex(db, txn));
+      p = p.next(() =>
+        this.createCollectionParentIndex(db, simpleDbTransaction)
+      );
     }
 
+    if (fromVersion < 9 && toVersion >= 9) {
+      p = p.next(() => {
+        createRemoteDocumentReadTimeIndex(txn);
+      });
+    }
     return p;
   }
 
@@ -294,6 +307,9 @@ export class DbTimestamp {
   constructor(public seconds: number, public nanoseconds: number) {}
 }
 
+/** A timestamp type that can be used in IndexedDb keys. */
+export type DbTimestampKey = number[];
+
 // The key for the singleton object in the DbPrimaryClient is a single string.
 export type DbPrimaryClientKey = typeof DbPrimaryClient.key;
 
@@ -590,6 +606,17 @@ export class DbUnknownDocument {
 export class DbRemoteDocument {
   static store = 'remoteDocuments';
 
+  /**
+   * An index that provides access to documents in a collection sorted by read
+   * time.
+   *
+   * This index is used to allow Index-Free queries to fetch newly changed
+   * documents in a collection.
+   */
+  static collectionReadTimeIndex = 'collectionReadTimeIndex';
+
+  static collectionReadTimeIndexPath = ['parentPath', 'readTime'];
+
   constructor(
     /**
      * Set to an instance of DbUnknownDocument if the data for a document is
@@ -613,7 +640,13 @@ export class DbRemoteDocument {
      * documents are potentially inconsistent with the backend's copy and use
      * the write's commit version as their document version.
      */
-    public hasCommittedMutations: boolean | undefined
+    public hasCommittedMutations: boolean | undefined,
+
+    /** When the document was read from the backend. */
+    public readTime: DbTimestampKey | undefined,
+
+    /** The path of the collection this document is part of. */
+    public parentPath: string | undefined
   ) {}
 }
 
@@ -957,6 +990,19 @@ function createRemoteDocumentChangesStore(db: IDBDatabase): void {
   });
 }
 
+/**
+ * Creates indices on the RemoteDocuments store used for both multi-tab
+ * and Index-Free queries.
+ */
+function createRemoteDocumentReadTimeIndex(txn: IDBTransaction): void {
+  const remoteDocumentStore = txn.objectStore(DbRemoteDocument.store);
+  remoteDocumentStore.createIndex(
+    DbRemoteDocument.collectionReadTimeIndex,
+    DbRemoteDocument.collectionReadTimeIndexPath,
+    { unique: false }
+  );
+}
+
 /**
  * A record of the metadata state of each client.
  *
@@ -1028,6 +1074,8 @@ export const V6_STORES = [...V4_STORES, DbRemoteDocumentGlobal.store];
 
 export const V8_STORES = [...V6_STORES, DbCollectionParent.store];
 
+// V9 does not change the set of stores.
+
 /**
  * The list of all default IndexedDB stores used throughout the SDK. This is
  * used when creating transactions so that access across all stores is done

packages/firestore/src/local/local_serializer.ts

@@ -39,6 +39,7 @@ import {
   DbRemoteDocument,
   DbTarget,
   DbTimestamp,
+  DbTimestampKey,
   DbUnknownDocument
 } from './indexeddb_schema';
 import { QueryData, QueryPurpose } from './query_data';
@@ -70,7 +71,12 @@ export class LocalSerializer {
   }
 
   /** Encodes a document for storage locally. */
-  toDbRemoteDocument(maybeDoc: MaybeDocument): DbRemoteDocument {
+  toDbRemoteDocument(
+    maybeDoc: MaybeDocument,
+    readTime: SnapshotVersion
+  ): DbRemoteDocument {
+    const dbReadTime = this.toDbTimestampKey(readTime);
+    const parentPath = maybeDoc.key.path.popLast().canonicalString();
     if (maybeDoc instanceof Document) {
       const doc = maybeDoc.proto
         ? maybeDoc.proto
@@ -80,7 +86,9 @@ export class LocalSerializer {
         /* unknownDocument= */ null,
         /* noDocument= */ null,
         doc,
-        hasCommittedMutations
+        hasCommittedMutations,
+        dbReadTime,
+        parentPath
       );
     } else if (maybeDoc instanceof NoDocument) {
       const path = maybeDoc.key.path.toArray();
@@ -90,7 +98,9 @@ export class LocalSerializer {
         /* unknownDocument= */ null,
         new DbNoDocument(path, readTime),
         /* document= */ null,
-        hasCommittedMutations
+        hasCommittedMutations,
+        dbReadTime,
+        parentPath
       );
     } else if (maybeDoc instanceof UnknownDocument) {
       const path = maybeDoc.key.path.toArray();
@@ -99,13 +109,20 @@ export class LocalSerializer {
         new DbUnknownDocument(path, readTime),
         /* noDocument= */ null,
         /* document= */ null,
-        /* hasCommittedMutations= */ true
+        /* hasCommittedMutations= */ true,
+        dbReadTime,
+        parentPath
       );
     } else {
       return fail('Unexpected MaybeDocumment');
     }
   }
 
+  toDbTimestampKey(snapshotVersion: SnapshotVersion): DbTimestampKey {
+    const timestamp = snapshotVersion.toTimestamp();
+    return [timestamp.seconds, timestamp.nanoseconds];
+  }
+
   private toDbTimestamp(snapshotVersion: SnapshotVersion): DbTimestamp {
     const timestamp = snapshotVersion.toTimestamp();
     return new DbTimestamp(timestamp.seconds, timestamp.nanoseconds);

packages/firestore/src/local/local_store.ts

@@ -351,7 +351,7 @@ export class LocalStore {
           .next(() =>
             this.applyWriteToRemoteDocuments(txn, batchResult, documentBuffer)
           )
-          .next(() => documentBuffer.apply(txn))
+          .next(() => documentBuffer.apply(txn, batchResult.commitVersion))
           .next(() => this.mutationQueue.performConsistencyCheck(txn))
           .next(() => this.localDocuments.getDocuments(txn, affected));
       }
@@ -449,7 +449,7 @@ export class LocalStore {
    */
   applyRemoteEvent(remoteEvent: RemoteEvent): Promise<MaybeDocumentMap> {
     const documentBuffer = this.remoteDocuments.newChangeBuffer();
-    const snapshotVersion = remoteEvent.snapshotVersion;
+    const remoteVersion = remoteEvent.snapshotVersion;
     return this.persistence.runTransaction(
       'Apply remote event',
       'readwrite-primary',
@@ -483,7 +483,7 @@ export class LocalStore {
             if (resumeToken.length > 0) {
               const newQueryData = oldQueryData.copy({
                 resumeToken,
-                snapshotVersion
+                snapshotVersion: remoteVersion
               });
               this.queryDataByTarget[targetId] = newQueryData;
 
@@ -561,28 +561,28 @@ export class LocalStore {
         // can synthesize remote events when we get permission denied errors while
         // trying to resolve the state of a locally cached document that is in
         // limbo.
-        if (!snapshotVersion.isEqual(SnapshotVersion.MIN)) {
+        if (!remoteVersion.isEqual(SnapshotVersion.MIN)) {
           const updateRemoteVersion = this.queryCache
             .getLastRemoteSnapshotVersion(txn)
             .next(lastRemoteSnapshotVersion => {
               assert(
-                snapshotVersion.compareTo(lastRemoteSnapshotVersion) >= 0,
+                remoteVersion.compareTo(lastRemoteSnapshotVersion) >= 0,
                 'Watch stream reverted to previous snapshot?? ' +
-                  snapshotVersion +
+                  remoteVersion +
                   ' < ' +
                   lastRemoteSnapshotVersion
               );
               return this.queryCache.setTargetsMetadata(
                 txn,
                 txn.currentSequenceNumber,
-                snapshotVersion
+                remoteVersion
               );
             });
           promises.push(updateRemoteVersion);
         }
 
         return PersistencePromise.waitFor(promises)
-          .next(() => documentBuffer.apply(txn))
+          .next(() => documentBuffer.apply(txn, remoteVersion))
           .next(() => {
             return this.localDocuments.getLocalViewOfDocuments(
               txn,

packages/firestore/src/local/memory_persistence.ts

@@ -49,6 +49,7 @@ import { PersistencePromise } from './persistence_promise';
 import { QueryData } from './query_data';
 import { ReferenceSet } from './reference_set';
 import { ClientId } from './shared_client_state';
+import { SnapshotVersion } from '../core/snapshot_version';
 
 const LOG_TAG = 'MemoryPersistence';
 
@@ -268,6 +269,7 @@ export class MemoryEagerDelegate implements ReferenceDelegate {
   onTransactionCommitted(
     txn: PersistenceTransaction
   ): PersistencePromise<void> {
+    // Remove newly orphaned documents.
     const cache = this.persistence.getRemoteDocumentCache();
     const changeBuffer = cache.newChangeBuffer();
     return PersistencePromise.forEach(
@@ -281,7 +283,7 @@ export class MemoryEagerDelegate implements ReferenceDelegate {
       }
     ).next(() => {
       this._orphanedDocuments = null;
-      return changeBuffer.apply(txn);
+      return changeBuffer.apply(txn, SnapshotVersion.forDeletedDoc());
     });
   }
 
@@ -419,7 +421,9 @@ export class MemoryLruDelegate implements ReferenceDelegate, LruDelegate {
         }
       });
     });
-    return p.next(() => changeBuffer.apply(txn)).next(() => count);
+    return p
+      .next(() => changeBuffer.apply(txn, SnapshotVersion.forDeletedDoc()))
+      .next(() => count);
   }
 
   removeMutationReference(
@@ -465,7 +469,10 @@ export class MemoryLruDelegate implements ReferenceDelegate, LruDelegate {
   }
 
   documentSize(maybeDoc: MaybeDocument): number {
-    const remoteDocument = this.serializer.toDbRemoteDocument(maybeDoc);
+    const remoteDocument = this.serializer.toDbRemoteDocument(
+      maybeDoc,
+      SnapshotVersion.forDeletedDoc()
+    );
     let value: unknown;
     if (remoteDocument.document) {
       value = remoteDocument.document;