Add basic tests

Author: wu-hui

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

@@ -65,13 +65,20 @@ import {
 } from '../lite-api/stage';
 import { Pipeline } from '../api/pipeline';
 import { Pipeline as LitePipeline } from '../lite-api/pipeline';
-import { canonifyQuery, Query, queryEquals, stringifyQuery } from './query';
+import {
+  canonifyQuery,
+  Query,
+  queryEquals,
+  QueryImpl,
+  stringifyQuery
+} from './query';
 import {
   canonifyTarget,
   Target,
   targetEquals,
   targetIsPipelineTarget
 } from './target';
+import { ResourcePath } from '../model/path';
 
 /* eslint @typescript-eslint/no-explicit-any: 0 */
 
@@ -405,7 +412,7 @@ export function getPipelineFlavor(p: Pipeline): PipelineFlavor {
 
 export type PipelineSourceType =
   | 'collection'
-  | 'collection-group'
+  | 'collection_group'
   | 'database'
   | 'documents';
 
@@ -416,10 +423,10 @@ export function getPipelineSourceType(
   const source = p.stages[0];
 
   if (
-    source.name === CollectionSource.name ||
-    source.name === CollectionGroupSource.name ||
-    source.name === DatabaseSource.name ||
-    source.name === DocumentsSource.name
+    source instanceof CollectionSource ||
+    source instanceof CollectionGroupSource ||
+    source instanceof DatabaseSource ||
+    source instanceof DocumentsSource
   ) {
     return source.name as PipelineSourceType;
   }
@@ -435,12 +442,34 @@ export function getPipelineCollection(p: Pipeline): string | undefined {
 }
 
 export function getPipelineCollectionGroup(p: Pipeline): string | undefined {
-  if (getPipelineSourceType(p) === 'collection-group') {
+  if (getPipelineSourceType(p) === 'collection_group') {
     return (p.stages[0] as CollectionGroupSource).collectionId;
   }
   return undefined;
 }
 
+export function asCollectionPipelineAtPath(
+  pipeline: Pipeline,
+  path: ResourcePath
+): Pipeline {
+  const newStages = pipeline.stages.map(s => {
+    if (s instanceof CollectionGroupSource) {
+      return new CollectionSource(path.canonicalString());
+    }
+
+    return s;
+  });
+
+  return new Pipeline(
+    pipeline.db,
+    pipeline.userDataReader,
+    pipeline.userDataWriter,
+    pipeline.documentReferenceFactory,
+    newStages,
+    pipeline.converter
+  );
+}
+
 export function getPipelineDocuments(p: Pipeline): string[] | undefined {
   if (getPipelineSourceType(p) === 'documents') {
     return (p.stages[0] as DocumentsSource).docPaths;
@@ -451,7 +480,7 @@ export function getPipelineDocuments(p: Pipeline): string[] | undefined {
 export type QueryOrPipeline = Query | Pipeline;
 
 export function isPipeline(q: QueryOrPipeline): q is Pipeline {
-  return q instanceof Pipeline;
+  return q instanceof Pipeline || q instanceof LitePipeline;
 }
 
 export function stringifyQueryOrPipeline(q: QueryOrPipeline): string {

packages/firestore/src/core/pipeline_run.ts

@@ -34,7 +34,7 @@ import {
 } from '../model/values';
 import { toEvaluable } from './expressions';
 import { UserDataReader } from '../lite-api/user_data_reader';
-import { Query, queryMatches } from './query';
+import { Query, queryMatches, queryMatchesAllDocuments } from './query';
 import { isPipeline, QueryOrPipeline } from './pipeline-util';
 
 export type PipelineInputOutput = MutableDocument;

packages/firestore/src/lite-api/pipeline.ts

@@ -150,7 +150,7 @@ export class Pipeline<AppModelType = DocumentData>
     readonly stages: Stage[],
     // TODO(pipeline) support converter
     //private converter:  FirestorePipelineConverter<AppModelType> = defaultPipelineConverter()
-    protected converter: unknown = {}
+    readonly converter: unknown = {}
   ) {}
 
   /**

packages/firestore/src/local/indexeddb_remote_document_cache.ts

@@ -59,6 +59,12 @@ import { QueryContext } from './query_context';
 import { RemoteDocumentCache } from './remote_document_cache';
 import { RemoteDocumentChangeBuffer } from './remote_document_change_buffer';
 import { SimpleDbStore } from './simple_db';
+import {
+  getPipelineCollection,
+  isPipeline,
+  QueryOrPipeline
+} from '../core/pipeline-util';
+import { queryOrPipelineMatches } from '../core/pipeline_run';
 
 export interface DocumentSizeEntry {
   document: MutableDocument;
@@ -199,7 +205,9 @@ class IndexedDbRemoteDocumentCacheImpl implements IndexedDbRemoteDocumentCache {
     return remoteDocumentsStore(transaction)
       .iterate((dbKey, dbDoc) => {
         const doc = this.maybeDecodeDocument(
-          DocumentKey.fromSegments(dbDoc.prefixPath.concat(dbDoc.documentId)),
+          DocumentKey.fromSegments(
+            dbDoc.prefixPath.concat(dbDoc.collectionGroup, dbDoc.documentId)
+          ),
           dbDoc
         );
         results = results.insert(doc.key, doc);
@@ -293,12 +301,21 @@ class IndexedDbRemoteDocumentCacheImpl implements IndexedDbRemoteDocumentCache {
 
   getDocumentsMatchingQuery(
     transaction: PersistenceTransaction,
-    query: Query,
+    query: QueryOrPipeline,
     offset: IndexOffset,
     mutatedDocs: OverlayMap,
     context?: QueryContext
   ): PersistencePromise<MutableDocumentMap> {
-    const collection = query.path;
+    if (isPipeline(query)) {
+      debugAssert(
+        !!getPipelineCollection(query),
+        'getDocumentsMatchingQuery can only handle collection pipelines'
+      );
+    }
+
+    const collection = isPipeline(query)
+      ? ResourcePath.fromString(getPipelineCollection(query)!)
+      : query.path;
     const startKey = [
       collection.popLast().toArray(),
       collection.lastSegment(),
@@ -331,7 +348,8 @@ class IndexedDbRemoteDocumentCacheImpl implements IndexedDbRemoteDocumentCache {
           );
           if (
             document.isFoundDocument() &&
-            (queryMatches(query, document) || mutatedDocs.has(document.key))
+            (queryOrPipelineMatches(query, document) ||
+              mutatedDocs.has(document.key))
           ) {
             // Either the document matches the given query, or it is mutated.
             results = results.insert(document.key, document);

packages/firestore/src/local/local_documents_view.ts

@@ -63,6 +63,7 @@ import { PersistenceTransaction } from './persistence_transaction';
 import { QueryContext } from './query_context';
 import { RemoteDocumentCache } from './remote_document_cache';
 import {
+  asCollectionPipelineAtPath,
   canonifyPipeline,
   getPipelineCollection,
   getPipelineCollectionGroup,
@@ -566,10 +567,33 @@ export class LocalDocumentsView {
     offset: IndexOffset,
     context?: QueryContext
   ): PersistencePromise<DocumentMap> {
-    if (getPipelineSourceType(pipeline) === 'collection-group') {
+    if (getPipelineSourceType(pipeline) === 'collection_group') {
       // TODO(pipeline): rewrite the pipeline as collection pipeline and recurse into this function
       // return this.getDocumentsMatchingPipeline(txn, pipeline, offset, context);
-      throw new Error('not implemented for collection group yet');
+      const collectionId = getPipelineCollectionGroup(pipeline)!;
+      let results = documentMap();
+      return this.indexManager
+        .getCollectionParents(txn, collectionId)
+        .next(parents => {
+          // Perform a collection query against each parent that contains the
+          // collectionId and aggregate the results.
+          return PersistencePromise.forEach(parents, (parent: ResourcePath) => {
+            const collectionPipeline = asCollectionPipelineAtPath(
+              pipeline,
+              parent.child(collectionId)
+            );
+            return this.getDocumentsMatchingPipeline(
+              txn,
+              collectionPipeline,
+              offset,
+              context
+            ).next(r => {
+              r.forEach((key, doc) => {
+                results = results.insert(key, doc);
+              });
+            });
+          }).next(() => results);
+        });
     } else {
       // Query the remote documents and overlay mutations.
       let overlays: OverlayMap;
@@ -661,7 +685,7 @@ export class LocalDocumentsView {
           ResourcePath.fromString(getPipelineCollection(pipeline)!),
           largestBatchId
         );
-      case 'collection-group':
+      case 'collection_group':
         throw new FirestoreError(
           'invalid-argument',
           `Unexpected collection group pipeline: ${canonifyPipeline(pipeline)}`

packages/firestore/src/local/query_engine.ts

@@ -59,7 +59,8 @@ import {
 import * as querystring from 'node:querystring';
 import {
   pipelineMatches,
-  pipelineMatchesAllDocuments
+  pipelineMatchesAllDocuments,
+  queryOrPipelineMatchesFullCollection
 } from '../core/pipeline_run';
 import { compareByKey } from '../model/document_comparator';
 

packages/firestore/test/integration/api/pipeline.listen.test.ts

@@ -262,7 +262,7 @@ apiDescribe('Pipelines', persistence => {
     ]);
   });
 
-  it.only('basic listen works', async () => {
+  it('basic listen works', async () => {
     const storeEvent = new EventsAccumulator<PipelineSnapshot>();
 
     let result = firestore

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

@@ -38,16 +38,9 @@ import { runPipeline } from '../../../src/core/pipeline_run';
 
 import { doc } from '../../util/helpers';
 import { and, or } from '../../../src/lite-api/expressions';
+import { newTestFirestore } from '../../util/api_helpers';
 
-const fakeAuthProvider: CredentialsProvider<User> =
-  {} as unknown as CredentialsProvider<User>;
-const fakeAppCheckProvider: CredentialsProvider<string> =
-  {} as unknown as CredentialsProvider<string>;
-const db = new Firestore(
-  fakeAuthProvider,
-  fakeAppCheckProvider,
-  DatabaseId.empty()
-);
+const db = newTestFirestore();
 
 describe('Pipeline Canonify', () => {
   it('works as expected for simple where clause', () => {
@@ -158,7 +151,7 @@ describe('Pipeline Canonify', () => {
   });
 });
 
-describe.only('pipelineEq', () => {
+describe('pipelineEq', () => {
   it('returns true for identical pipelines', () => {
     const p1 = db.pipeline().collection('test').where(eq(`foo`, 42));
     const p2 = db.pipeline().collection('test').where(eq(`foo`, 42));