Merge remote-tracking branch 'origin/mila/BloomFilter' into BloomFilterComplexIntegrationTest

Author: dconeybe

packages/firestore/src/core/sync_engine_impl.ts

@@ -71,7 +71,6 @@ import { primitiveComparator } from '../util/misc';
 import { ObjectMap } from '../util/obj_map';
 import { Deferred } from '../util/promise';
 import { SortedMap } from '../util/sorted_map';
-import { SortedSet } from '../util/sorted_set';
 import { BATCHID_UNKNOWN } from '../util/types';
 
 import {
@@ -639,7 +638,9 @@ export async function syncEngineRejectListen(
     const event = new RemoteEvent(
       SnapshotVersion.min(),
       /* targetChanges= */ new Map<TargetId, TargetChange>(),
-      /* targetMismatches= */ new SortedSet<TargetId>(primitiveComparator),
+      /* targetMismatches= */ new SortedMap<TargetId, TargetPurpose>(
+        primitiveComparator
+      ),
       documentUpdates,
       resolvedLimboDocuments
     );

packages/firestore/src/local/local_store_impl.ts

@@ -601,7 +601,7 @@ export function localStoreApplyRemoteEventToLocalCache(
         let newTargetData = oldTargetData.withSequenceNumber(
           txn.currentSequenceNumber
         );
-        if (remoteEvent.targetMismatches.has(targetId)) {
+        if (remoteEvent.targetMismatches.get(targetId) !== null) {
           newTargetData = newTargetData
             .withResumeToken(
               ByteString.EMPTY_BYTE_STRING,

packages/firestore/src/local/target_data.ts

@@ -26,10 +26,17 @@ export const enum TargetPurpose {
   Listen,
 
   /**
-   * The query target was used to refill a query after an existence filter mismatch.
+   * The query target was used to refill a query after an existence filter
+   * mismatch.
    */
   ExistenceFilterMismatch,
 
+  /**
+   * The query target was used if the query is the result of a false positive in
+   * the bloom filter.
+   */
+  ExistenceFilterMismatchBloom,
+
   /** The query target was used to resolve a limbo document. */
   LimboResolution
 }

packages/firestore/src/remote/remote_event.ts

@@ -17,15 +17,16 @@
 
 import { SnapshotVersion } from '../core/snapshot_version';
 import { TargetId } from '../core/types';
+import { TargetPurpose } from '../local/target_data';
 import {
   documentKeySet,
   DocumentKeySet,
   mutableDocumentMap,
-  MutableDocumentMap,
-  targetIdSet
+  MutableDocumentMap
 } from '../model/collections';
 import { ByteString } from '../util/byte_string';
-import { SortedSet } from '../util/sorted_set';
+import { primitiveComparator } from '../util/misc';
+import { SortedMap } from '../util/sorted_map';
 
 /**
  * An event from the RemoteStore. It is split into targetChanges (changes to the
@@ -43,10 +44,11 @@ export class RemoteEvent {
      */
     readonly targetChanges: Map<TargetId, TargetChange>,
     /**
-     * A set of targets that is known to be inconsistent. Listens for these
-     * targets should be re-established without resume tokens.
+     * A map of targets that is known to be inconsistent, and the purpose for
+     * re-listening. Listens for these targets should be re-established without
+     * resume tokens.
      */
-    readonly targetMismatches: SortedSet<TargetId>,
+    readonly targetMismatches: SortedMap<TargetId, TargetPurpose>,
     /**
      * A set of which documents have changed or been deleted, along with the
      * doc's new values (if not deleted).
@@ -82,7 +84,7 @@ export class RemoteEvent {
     return new RemoteEvent(
       SnapshotVersion.min(),
       targetChanges,
-      targetIdSet(),
+      new SortedMap<TargetId, TargetPurpose>(primitiveComparator),
       mutableDocumentMap(),
       documentKeySet()
     );

packages/firestore/src/remote/remote_store.ts

@@ -24,7 +24,7 @@ import {
   localStoreGetNextMutationBatch
 } from '../local/local_store_impl';
 import { isIndexedDbTransactionError } from '../local/simple_db';
-import { TargetData, TargetPurpose } from '../local/target_data';
+import { TargetData } from '../local/target_data';
 import { MutationResult } from '../model/mutation';
 import { MutationBatch, MutationBatchResult } from '../model/mutation_batch';
 import { debugAssert, debugCast } from '../util/assert';
@@ -587,7 +587,7 @@ function raiseWatchSnapshot(
 
   // Re-establish listens for the targets that have been invalidated by
   // existence filter mismatches.
-  remoteEvent.targetMismatches.forEach(targetId => {
+  remoteEvent.targetMismatches.forEach((targetId, targetPurpose) => {
     const targetData = remoteStoreImpl.listenTargets.get(targetId);
     if (!targetData) {
       // A watched target might have been removed already.
@@ -615,7 +615,7 @@ function raiseWatchSnapshot(
     const requestTargetData = new TargetData(
       targetData.target,
       targetId,
-      TargetPurpose.ExistenceFilterMismatch,
+      targetPurpose,
       targetData.sequenceNumber
     );
     sendWatchRequest(remoteStoreImpl, requestTargetData);

packages/firestore/src/remote/serializer.ts

@@ -1008,7 +1008,7 @@ export function toListenRequestLabels(
   serializer: JsonProtoSerializer,
   targetData: TargetData
 ): ProtoApiClientObjectMap<string> | null {
-  const value = toLabel(serializer, targetData.purpose);
+  const value = toLabel(targetData.purpose);
   if (value == null) {
     return null;
   } else {
@@ -1018,15 +1018,14 @@ export function toListenRequestLabels(
   }
 }
 
-function toLabel(
-  serializer: JsonProtoSerializer,
-  purpose: TargetPurpose
-): string | null {
+export function toLabel(purpose: TargetPurpose): string | null {
   switch (purpose) {
     case TargetPurpose.Listen:
       return null;
     case TargetPurpose.ExistenceFilterMismatch:
       return 'existence-filter-mismatch';
+    case TargetPurpose.ExistenceFilterMismatchBloom:
+      return 'existence-filter-mismatch-bloom';
     case TargetPurpose.LimboResolution:
       return 'limbo-document';
     default:

packages/firestore/src/remote/watch_change.ts

@@ -91,6 +91,11 @@ export const enum WatchTargetChangeState {
   Reset
 }
 
+const enum BloomFilterApplicationStatus {
+  Success,
+  Skipped,
+  FalsePositive
+}
 export class WatchTargetChange {
   constructor(
     /** What kind of change occurred to the watch target. */
@@ -284,11 +289,13 @@ export class WatchChangeAggregator {
   private pendingDocumentTargetMapping = documentTargetMap();
 
   /**
-   * A list of targets with existence filter mismatches. These targets are
+   * A map of targets with existence filter mismatches. These targets are
    * known to be inconsistent and their listens needs to be re-established by
    * RemoteStore.
    */
-  private pendingTargetResets = new SortedSet<TargetId>(primitiveComparator);
+  private pendingTargetResets = new SortedMap<TargetId, TargetPurpose>(
+    primitiveComparator
+  );
 
   /**
    * Processes and adds the DocumentWatchChange to the current set of changes.
@@ -426,15 +433,21 @@ export class WatchChangeAggregator {
         // raise a snapshot with `isFromCache:true`.
         if (currentSize !== expectedCount) {
           // Apply bloom filter to identify and mark removed documents.
-          const bloomFilterApplied = this.applyBloomFilter(
-            watchChange,
-            currentSize
-          );
-          if (!bloomFilterApplied) {
+          const status = this.applyBloomFilter(watchChange, currentSize);
+
+          if (status !== BloomFilterApplicationStatus.Success) {
             // If bloom filter application fails, we reset the mapping and
             // trigger re-run of the query.
             this.resetTarget(targetId);
-            this.pendingTargetResets = this.pendingTargetResets.add(targetId);
+
+            const purpose: TargetPurpose =
+              status === BloomFilterApplicationStatus.FalsePositive
+                ? TargetPurpose.ExistenceFilterMismatchBloom
+                : TargetPurpose.ExistenceFilterMismatch;
+            this.pendingTargetResets = this.pendingTargetResets.insert(
+              targetId,
+              purpose
+            );
           }
           TestingHooks.instance?.notifyOnExistenceFilterMismatch(
             createExistenceFilterMismatchInfoForTestingHooks(
@@ -448,16 +461,19 @@ export class WatchChangeAggregator {
     }
   }
 
-  /** Returns whether a bloom filter removed the deleted documents successfully. */
+  /**
+   * Apply bloom filter to remove the deleted documents, and return the
+   * application status.
+   */
   private applyBloomFilter(
     watchChange: ExistenceFilterChange,
     currentCount: number
-  ): boolean {
+  ): BloomFilterApplicationStatus {
     const { unchangedNames, count: expectedCount } =
       watchChange.existenceFilter;
 
     if (!unchangedNames || !unchangedNames.bits) {
-      return false;
+      return BloomFilterApplicationStatus.Skipped;
     }
 
     const {
@@ -475,7 +491,7 @@ export class WatchChangeAggregator {
             err.message +
             '); ignoring the bloom filter and falling back to full re-query.'
         );
-        return false;
+        return BloomFilterApplicationStatus.Skipped;
       } else {
         throw err;
       }
@@ -491,15 +507,23 @@ export class WatchChangeAggregator {
       } else {
         logWarn('Applying bloom filter failed: ', err);
       }
-      return false;
+      return BloomFilterApplicationStatus.Skipped;
+    }
+
+    if (bloomFilter.bitCount === 0) {
+      return BloomFilterApplicationStatus.Skipped;
     }
 
     const removedDocumentCount = this.filterRemovedDocuments(
       watchChange.targetId,
       bloomFilter
     );
 
-    return expectedCount === currentCount - removedDocumentCount;
+    if (expectedCount !== currentCount - removedDocumentCount) {
+      return BloomFilterApplicationStatus.FalsePositive;
+    }
+
+    return BloomFilterApplicationStatus.Success;
   }
 
   /**
@@ -610,7 +634,9 @@ export class WatchChangeAggregator {
 
     this.pendingDocumentUpdates = mutableDocumentMap();
     this.pendingDocumentTargetMapping = documentTargetMap();
-    this.pendingTargetResets = new SortedSet<TargetId>(primitiveComparator);
+    this.pendingTargetResets = new SortedMap<TargetId, TargetPurpose>(
+      primitiveComparator
+    );
 
     return remoteEvent;
   }

packages/firestore/test/unit/remote/remote_event.test.ts

@@ -431,6 +431,9 @@ describe('RemoteEvent', () => {
     expectTargetChangeEquals(event.targetChanges.get(1)!, expected);
   });
 
+  // TODO(b/272564458): Add test cases for existence filter with bloom filter,
+  // one will skip the re-query, one will yield false positive result and clears
+  // target mapping.
   it('existence filters clears target mapping', () => {
     const targets = listens(1, 2);
 
@@ -459,6 +462,9 @@ describe('RemoteEvent', () => {
     event = aggregator.createRemoteEvent(version(3));
     expect(event.documentUpdates.size).to.equal(0);
     expect(event.targetMismatches.size).to.equal(1);
+    expect(event.targetMismatches.get(1)).to.equal(
+      TargetPurpose.ExistenceFilterMismatch
+    );
     expect(event.targetChanges.size).to.equal(1);
 
     const expected = updateMapping(
@@ -499,6 +505,9 @@ describe('RemoteEvent', () => {
     const event = aggregator.createRemoteEvent(version(3));
     expect(event.documentUpdates.size).to.equal(1);
     expect(event.targetMismatches.size).to.equal(1);
+    expect(event.targetMismatches.get(1)).to.equal(
+      TargetPurpose.ExistenceFilterMismatch
+    );
     expect(event.targetChanges.get(1)!.current).to.be.false;
   });
 

packages/firestore/test/unit/specs/existence_filter_spec.test.ts

@@ -16,6 +16,7 @@
  */
 
 import { newQueryForPath } from '../../../src/core/query';
+import { TargetPurpose } from '../../../src/local/target_data';
 import { Code } from '../../../src/util/error';
 import {
   deletedDoc,
@@ -305,7 +306,11 @@ describeSpec('Existence Filters:', [], () => {
           // BloomFilter correctly identifies docC is deleted, but yields false
           // positive results for docB. Re-run query is triggered.
           .expectEvents(query1, { fromCache: true })
-          .expectActiveTargets({ query: query1, resumeToken: '' })
+          .expectActiveTargets({
+            query: query1,
+            resumeToken: '',
+            targetPurpose: TargetPurpose.ExistenceFilterMismatchBloom
+          })
       );
     }
   );
@@ -394,7 +399,11 @@ describeSpec('Existence Filters:', [], () => {
           .watchSnapshots(2000)
           // Re-run query is triggered.
           .expectEvents(query1, { fromCache: true })
-          .expectActiveTargets({ query: query1, resumeToken: '' })
+          .expectActiveTargets({
+            query: query1,
+            resumeToken: '',
+            targetPurpose: TargetPurpose.ExistenceFilterMismatch
+          })
       );
     }
   );
@@ -424,7 +433,11 @@ describeSpec('Existence Filters:', [], () => {
           .watchSnapshots(2000)
           // Re-run query is triggered.
           .expectEvents(query1, { fromCache: true })
-          .expectActiveTargets({ query: query1, resumeToken: '' })
+          .expectActiveTargets({
+            query: query1,
+            resumeToken: '',
+            targetPurpose: TargetPurpose.ExistenceFilterMismatch
+          })
       );
     }
   );
@@ -452,7 +465,11 @@ describeSpec('Existence Filters:', [], () => {
         .watchSnapshots(2000)
         // Re-run query is triggered.
         .expectEvents(query1, { fromCache: true })
-        .expectActiveTargets({ query: query1, resumeToken: '' })
+        .expectActiveTargets({
+          query: query1,
+          resumeToken: '',
+          targetPurpose: TargetPurpose.ExistenceFilterMismatch
+        })
     );
   });