memory spec tests pass sans limitToLast

Author: wu-hui

packages/firestore/src/core/pipeline-util.ts

@@ -349,7 +349,8 @@ function canonifyExpr(expr: Expr): string {
     return `fld(${expr.fieldName()})`;
   }
   if (expr instanceof Constant) {
-    return `cst(${expr.value})`;
+    // TODO(pipeline): use better alternatives than JSON.stringify
+    return `cst(${JSON.stringify(expr.value)})`;
   }
   if (expr instanceof FirestoreFunction) {
     return `fn(${expr.name},[${expr.params.map(canonifyExpr).join(',')}])`;

packages/firestore/src/core/sync_engine_impl.ts

@@ -930,14 +930,6 @@ function removeAndCleanupTarget(
 ): void {
   syncEngineImpl.sharedClientState.removeLocalQueryTarget(targetId);
 
-  // TODO(pipeline): REMOVE this hack.
-  if (
-    !syncEngineImpl.queriesByTarget.has(targetId) ||
-    syncEngineImpl.queriesByTarget.get(targetId)!.length !== 0
-  ) {
-    return;
-  }
-
   debugAssert(
     syncEngineImpl.queriesByTarget.has(targetId) &&
       syncEngineImpl.queriesByTarget.get(targetId)!.length !== 0,

packages/firestore/src/remote/watch_change.ts

@@ -45,6 +45,15 @@ import {
 import { BloomFilter, BloomFilterError } from './bloom_filter';
 import { ExistenceFilter } from './existence_filter';
 import { RemoteEvent, TargetChange } from './remote_event';
+import {
+  getPipelineDocuments,
+  getPipelineFlavor,
+  getPipelineSourceType,
+  isPipeline,
+  TargetOrPipeline
+} from '../core/pipeline-util';
+import { Pipeline } from '../lite-api/pipeline';
+import { ResourcePath } from '../model/path';
 
 /**
  * Internal representation of the watcher API protocol buffers.
@@ -405,6 +414,17 @@ export class WatchChangeAggregator {
     }
   }
 
+  isSingleDocumentTarget(target: TargetOrPipeline): boolean {
+    if (targetIsPipelineTarget(target)) {
+      return (
+        getPipelineSourceType(target) === 'documents' &&
+        getPipelineDocuments(target)?.length === 1
+      );
+    }
+
+    return targetIsDocumentTarget(target);
+  }
+
   /**
    * Handles existence filters and synthesizes deletes for filter mismatches.
    * Targets that are invalidated by filter mismatches are added to
@@ -417,29 +437,7 @@ export class WatchChangeAggregator {
     const targetData = this.targetDataForActiveTarget(targetId);
     if (targetData) {
       const target = targetData.target;
-      if (targetIsPipelineTarget(target)) {
-        //TODO(pipeline): handle existence filter correctly for pipelines
-      } else if (targetIsDocumentTarget(target)) {
-        if (expectedCount === 0) {
-          // The existence filter told us the document does not exist. We deduce
-          // that this document does not exist and apply a deleted document to
-          // our updates. Without applying this deleted document there might be
-          // another query that will raise this document as part of a snapshot
-          // until it is resolved, essentially exposing inconsistency between
-          // queries.
-          const key = new DocumentKey(target.path);
-          this.removeDocumentFromTarget(
-            targetId,
-            key,
-            MutableDocument.newNoDocument(key, SnapshotVersion.min())
-          );
-        } else {
-          hardAssert(
-            expectedCount === 1,
-            'Single document existence filter with count: ' + expectedCount
-          );
-        }
-      } else {
+      if (!this.isSingleDocumentTarget(target)) {
         const currentSize = this.getCurrentDocumentCountForTarget(targetId);
         // Existence filter mismatch. Mark the documents as being in limbo, and
         // raise a snapshot with `isFromCache:true`.
@@ -474,6 +472,30 @@ export class WatchChangeAggregator {
             )
           );
         }
+      } else {
+        if (expectedCount === 0) {
+          // The existence filter told us the document does not exist. We deduce
+          // that this document does not exist and apply a deleted document to
+          // our updates. Without applying this deleted document there might be
+          // another query that will raise this document as part of a snapshot
+          // until it is resolved, essentially exposing inconsistency between
+          // queries.
+          const key = new DocumentKey(
+            targetIsPipelineTarget(target)
+              ? ResourcePath.fromString(getPipelineDocuments(target)![0])
+              : target.path
+          );
+          this.removeDocumentFromTarget(
+            targetId,
+            key,
+            MutableDocument.newNoDocument(key, SnapshotVersion.min())
+          );
+        } else {
+          hardAssert(
+            expectedCount === 1,
+            'Single document existence filter with count: ' + expectedCount
+          );
+        }
       }
     }
   }
@@ -591,8 +613,7 @@ export class WatchChangeAggregator {
       if (targetData) {
         if (
           targetState.current &&
-          !targetIsPipelineTarget(targetData.target) &&
-          targetIsDocumentTarget(targetData.target)
+          this.isSingleDocumentTarget(targetData.target)
         ) {
           // Document queries for document that don't exist can produce an empty
           // result set. To update our local cache, we synthesize a document
@@ -603,7 +624,12 @@ export class WatchChangeAggregator {
           // TODO(dimond): Ideally we would have an explicit lookup target
           // instead resulting in an explicit delete message and we could
           // remove this special logic.
-          const key = new DocumentKey(targetData.target.path);
+          const path = targetIsPipelineTarget(targetData.target)
+            ? ResourcePath.fromString(
+                getPipelineDocuments(targetData.target)![0]
+              )
+            : targetData.target.path;
+          const key = new DocumentKey(path);
           if (
             this.pendingDocumentUpdates.get(key) === null &&
             !this.targetContainsDocument(targetId, key)
@@ -695,7 +721,12 @@ export class WatchChangeAggregator {
     targetState.addDocumentChange(document.key, changeType);
 
     if (
-      targetIsPipelineTarget(this.targetDataForActiveTarget(targetId)!.target)
+      targetIsPipelineTarget(
+        this.targetDataForActiveTarget(targetId)!.target
+      ) &&
+      getPipelineFlavor(
+        this.targetDataForActiveTarget(targetId)!.target as Pipeline
+      ) !== 'exact'
     ) {
       this.pendingAugmentedDocumentUpdates =
         this.pendingAugmentedDocumentUpdates.insert(document.key, document);
@@ -747,7 +778,12 @@ export class WatchChangeAggregator {
 
     if (updatedDocument) {
       if (
-        targetIsPipelineTarget(this.targetDataForActiveTarget(targetId)!.target)
+        targetIsPipelineTarget(
+          this.targetDataForActiveTarget(targetId)!.target
+        ) &&
+        getPipelineFlavor(
+          this.targetDataForActiveTarget(targetId)!.target as Pipeline
+        ) !== 'exact'
       ) {
         this.pendingAugmentedDocumentUpdates =
           this.pendingAugmentedDocumentUpdates.insert(key, updatedDocument);

packages/firestore/test/unit/core/pipeline.test.ts

@@ -199,7 +199,7 @@ describe('pipelineEq', () => {
   });
 });
 
-describe.only('runPipeline()', () => {
+describe('runPipeline()', () => {
   it('works with collection stage', () => {
     const p = db.pipeline().collection('test');
 

packages/firestore/test/unit/local/local_store.test.ts

@@ -596,7 +596,7 @@ function compareDocsWithCreateTime(
   );
 }
 
-describe.only('LocalStore w/ Memory Persistence', () => {
+describe('LocalStore w/ Memory Persistence', () => {
   async function initialize(): Promise<LocalStoreComponents> {
     const queryEngine = new CountingQueryEngine();
     const persistence = await persistenceHelpers.testMemoryEagerPersistence();
@@ -616,7 +616,7 @@ describe.only('LocalStore w/ Memory Persistence', () => {
   });
 });
 
-describe.only('LocalStore w/ Memory Persistence and Pipelines', () => {
+describe('LocalStore w/ Memory Persistence and Pipelines', () => {
   async function initialize(): Promise<LocalStoreComponents> {
     const queryEngine = new CountingQueryEngine();
     const persistence = await persistenceHelpers.testMemoryEagerPersistence();
@@ -663,7 +663,7 @@ describe('LocalStore w/ IndexedDB Persistence', () => {
   });
 });
 
-describe.only('LocalStore w/ IndexedDB Persistence and Pipeline', () => {
+describe('LocalStore w/ IndexedDB Persistence and Pipeline', () => {
   if (!IndexedDbPersistence.isAvailable()) {
     console.warn(
       'No IndexedDB. Skipping LocalStore w/ IndexedDB persistence tests.'
@@ -2507,7 +2507,7 @@ function genericLocalStoreTests(
   });
 
   // eslint-disable-next-line no-restricted-properties
-  (options.gcIsEager ? it.skip : it.only)(
+  (options.gcIsEager ? it.skip : it)(
     'ignores target mapping after existence filter mismatch',
     async () => {
       const query1 = query('foo', filter('matches', '==', true));

packages/firestore/test/unit/local/query_engine.test.ts

@@ -123,7 +123,7 @@ class TestLocalDocumentsView extends LocalDocumentsView {
   }
 }
 
-describe.only('QueryEngine', async () => {
+describe('QueryEngine', async () => {
   describe('MemoryEagerPersistence usePipeline=false', async () => {
     /* not durable and without client side indexing */
     genericQueryEngineTest(persistenceHelpers.testMemoryEagerPersistence, {

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

@@ -36,6 +36,7 @@ import { TestBundleBuilder } from '../util/bundle_data';
 
 import { describeSpec, specTest } from './describe_spec';
 import { client, spec } from './spec_builder';
+import { setLogLevel } from '../../../src';
 
 interface TestBundleDocument {
   key: DocumentKey;
@@ -285,32 +286,36 @@ describeSpec('Bundles:', [], () => {
     );
   });
 
-  specTest('Bundles query can be resumed from same query.', [], () => {
-    const query1 = query('collection');
-    const docA = doc('collection/a', 100, { key: 'a' });
-    const bundleString1 = bundleWithDocumentAndQuery(
-      {
-        key: docA.key,
-        readTime: 500,
-        createTime: 250,
-        updateTime: 500,
-        content: { value: 'b' }
-      },
-      { name: 'bundled-query', readTime: 400, query: query1 }
-    );
+  specTest(
+    'Bundles query can be resumed from same query.',
+    ['no-pipeline-conversion'],
+    () => {
+      const query1 = query('collection');
+      const docA = doc('collection/a', 100, { key: 'a' });
+      const bundleString1 = bundleWithDocumentAndQuery(
+        {
+          key: docA.key,
+          readTime: 500,
+          createTime: 250,
+          updateTime: 500,
+          content: { value: 'b' }
+        },
+        { name: 'bundled-query', readTime: 400, query: query1 }
+      );
 
-    return spec()
-      .loadBundle(bundleString1)
-      .userListens(query1, { readTime: 400 })
-      .expectEvents(query1, {
-        added: [doc('collection/a', 500, { value: 'b' })],
-        fromCache: true
-      });
-  });
+      return spec()
+        .loadBundle(bundleString1)
+        .userListens(query1, { readTime: 400 })
+        .expectEvents(query1, {
+          added: [doc('collection/a', 500, { value: 'b' })],
+          fromCache: true
+        });
+    }
+  );
 
   specTest(
     'Bundles query can be loaded and resumed from different tabs',
-    ['multi-client'],
+    ['multi-client', 'no-pipeline-conversion'],
     () => {
       const query1 = query('collection');
       const query2 = query('collection', filter('value', '==', 'c'));