limit to last, cursors and multitab for documents and database stages

Author: wu-hui

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

@@ -13,12 +13,17 @@
 // limitations under the License.
 
 import {
+  And,
   and,
   Constant,
   Expr,
   Field,
   FilterCondition,
   FirestoreFunction,
+  gt,
+  gte,
+  lt,
+  lte,
   not,
   or,
   Ordering
@@ -85,6 +90,7 @@ import { Firestore } from '../api/database';
 import { doc } from '../lite-api/reference';
 import { Direction } from './order_by';
 import { CorePipeline } from './pipeline_run';
+import { Bound } from './bound';
 
 /* eslint @typescript-eslint/no-explicit-any: 0 */
 
@@ -295,6 +301,16 @@ export function toPipelineFilterCondition(
   throw new Error(`Failed to convert filter to pipeline conditions: ${f}`);
 }
 
+function reverseOrderings(orderings: Ordering[]): Ordering[] {
+  return orderings.map(
+    o =>
+      new Ordering(
+        o.expr,
+        o.direction === 'ascending' ? 'descending' : 'ascending'
+      )
+  );
+}
+
 export function toPipeline(query: Query, db: Firestore): Pipeline {
   let pipeline: Pipeline;
   if (isCollectionGroupQuery(query)) {
@@ -323,28 +339,68 @@ export function toPipeline(query: Query, db: Firestore): Pipeline {
     pipeline = pipeline.where(existsConditions[0]);
   }
 
-  pipeline = pipeline.sort(
-    ...orders.map(order =>
-      order.dir === Direction.ASCENDING
-        ? Field.of(order.field.canonicalString()).ascending()
-        : Field.of(order.field.canonicalString()).descending()
-    )
+  const orderings = orders.map(order =>
+    order.dir === Direction.ASCENDING
+      ? Field.of(order.field.canonicalString()).ascending()
+      : Field.of(order.field.canonicalString()).descending()
   );
 
-  // cursors and limits
-  if (query.startAt !== null || query.endAt !== null) {
-    throw new Error('Cursors are not supported yet.');
-  }
   if (query.limitType === LimitType.Last) {
-    throw new Error('Limit to last are not supported yet.');
-  }
-  if (query.limit !== null) {
-    pipeline = pipeline.limit(query.limit);
+    pipeline = pipeline.sort(...reverseOrderings(orderings));
+    // cursors
+    if (query.startAt !== null) {
+      pipeline = pipeline.where(
+        whereConditionsFromCursor(query.startAt, orderings, 'before')
+      );
+    }
+
+    if (query.endAt !== null) {
+      pipeline = pipeline.where(
+        whereConditionsFromCursor(query.endAt, orderings, 'after')
+      );
+    }
+
+    pipeline = pipeline._limit(query.limit!, true);
+    pipeline = pipeline.sort(...orderings);
+  } else {
+    pipeline = pipeline.sort(...orderings);
+    if (query.startAt !== null) {
+      pipeline = pipeline.where(
+        whereConditionsFromCursor(query.startAt, orderings, 'after')
+      );
+    }
+    if (query.endAt !== null) {
+      pipeline = pipeline.where(
+        whereConditionsFromCursor(query.endAt, orderings, 'before')
+      );
+    }
+
+    if (query.limit !== null) {
+      pipeline = pipeline.limit(query.limit);
+    }
   }
 
   return pipeline;
 }
 
+function whereConditionsFromCursor(
+  bound: Bound,
+  orderings: Ordering[],
+  position: 'before' | 'after'
+): And {
+  const cursors = bound.position.map(value => Constant._fromProto(value));
+  const filterFunc = position === 'before' ? lt : gt;
+  const filterInclusiveFunc = position === 'before' ? lte : gte;
+  const conditions = cursors.map((cursor, index) => {
+    if (!!bound.inclusive && index === cursors.length - 1) {
+      return filterInclusiveFunc(orderings[index].expr as Field, cursor);
+    } else {
+      return filterFunc(orderings[index].expr as Field, cursor);
+    }
+  });
+  return new And(conditions);
+}
+
 function canonifyExpr(expr: Expr): string {
   if (expr instanceof Field) {
     return `fld(${expr.fieldName()})`;

packages/firestore/src/core/pipeline_run.ts

@@ -283,15 +283,16 @@ function lastEffectiveSort(pipeline: CorePipeline): Ordering[] {
 
 export function getLastEffectiveLimit(
   pipeline: CorePipeline
-): number | undefined {
-  // return the last sort stage, throws exception if it doesn't exist
-  // TODO(pipeline): this implementation is wrong, there are stages that can invalidate
-  // the orderings later. The proper way to manipulate the pipeline so that last Sort
-  // always has effects.
+): { limit: number; convertedFromLimitToLast: boolean } | undefined {
+  // TODO(pipeline): this implementation is wrong, there are stages that can change
+  // the limit later (findNearest).
   for (let i = pipeline.stages.length - 1; i >= 0; i--) {
     const stage = pipeline.stages[i];
     if (stage instanceof Limit) {
-      return stage.limit;
+      return {
+        limit: stage.limit,
+        convertedFromLimitToLast: stage.convertedFromLimitTolast
+      };
     }
   }
   return undefined;

packages/firestore/src/core/sync_engine_impl.ts

@@ -25,6 +25,7 @@ import {
   localStoreExecuteQuery,
   localStoreGetActiveClients,
   localStoreGetCachedTarget,
+  localStoreGetDocuments,
   localStoreGetHighestUnacknowledgedBatchId,
   localStoreGetNewDocumentChanges,
   localStoreHandleUserChange,
@@ -45,6 +46,7 @@ import { TargetData, TargetPurpose } from '../local/target_data';
 import {
   DocumentKeySet,
   documentKeySet,
+  documentMap,
   DocumentMap,
   mutableDocumentMap
 } from '../model/collections';
@@ -120,6 +122,7 @@ import {
   canonifyQueryOrPipeline,
   getPipelineCollection,
   getPipelineCollectionId,
+  getPipelineSourceType,
   isPipeline,
   QueryOrPipeline,
   queryOrPipelineEqual,
@@ -1096,7 +1099,6 @@ export async function syncEngineEmitNewSnapsAndNotifyLocalStore(
     return;
   }
 
-  // TODO(pipeline): will this work for pipelines?
   syncEngineImpl.queryViewsByQuery.forEach((_, queryView) => {
     debugAssert(
       !!syncEngineImpl.applyDocChanges,
@@ -1548,13 +1550,35 @@ export async function syncEngineApplyTargetState(
     switch (state) {
       case 'current':
       case 'not-current': {
-        const changes = await localStoreGetNewDocumentChanges(
-          syncEngineImpl.localStore,
-          // TODO(pipeline): handle database/documents pipeline
-          isPipeline(query[0])
-            ? getPipelineCollectionId(query[0])!
-            : queryCollectionGroup(query[0])
-        );
+        let changes: DocumentMap;
+        if (isPipeline(query[0])) {
+          switch (getPipelineSourceType(query[0])) {
+            case 'collection_group':
+            case 'collection':
+              changes = await localStoreGetNewDocumentChanges(
+                syncEngineImpl.localStore,
+                getPipelineCollectionId(query[0])!
+              );
+              break;
+            case 'documents':
+              changes = await localStoreGetDocuments(
+                syncEngineImpl.localStore,
+                query[0]!
+              );
+              break;
+            case 'database':
+            case 'unknown':
+              logWarn('');
+              changes = documentMap();
+              break;
+          }
+        } else {
+          changes = await localStoreGetNewDocumentChanges(
+            syncEngineImpl.localStore,
+            queryCollectionGroup(query[0])
+          );
+        }
+
         const synthesizedRemoteEvent =
           RemoteEvent.createSynthesizedRemoteEventForCurrentChange(
             targetId,

packages/firestore/src/core/view.ts

@@ -252,19 +252,25 @@ export class View {
 
   private getLimit(query: QueryOrPipeline): number | undefined {
     return isPipeline(query)
-      ? getLastEffectiveLimit(query)
+      ? getLastEffectiveLimit(query)?.limit
       : query.limit || undefined;
   }
+
   private getLimitType(query: QueryOrPipeline): LimitType {
-    return isPipeline(query) ? LimitType.First : query.limitType;
+    return isPipeline(query)
+      ? getLastEffectiveLimit(query)?.convertedFromLimitToLast
+        ? LimitType.Last
+        : LimitType.First
+      : query.limitType;
+    // return isPipeline(query) ? LimitType.First : query.limitType;
   }
 
   private getLimitEdges(
     query: QueryOrPipeline,
     oldDocumentSet: DocumentSet
   ): [Document | null, Document | null] {
     if (isPipeline(query)) {
-      const limit = getLastEffectiveLimit(query);
+      const limit = getLastEffectiveLimit(query)?.limit;
       return [
         oldDocumentSet.size === limit ? oldDocumentSet.last() : null,
         null

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

@@ -405,6 +405,22 @@ export class Pipeline<AppModelType = DocumentData>
     );
   }
 
+  _limit(
+    limit: number,
+    convertedFromLimitTolast: boolean
+  ): Pipeline<AppModelType> {
+    const copy = this.stages.map(s => s);
+    copy.push(new Limit(limit, convertedFromLimitTolast));
+    return new Pipeline(
+      this.liteDb,
+      this.userDataReader,
+      this.userDataWriter,
+      this.documentReferenceFactory,
+      copy,
+      this.converter
+    );
+  }
+
   /**
    * Returns a set of distinct {@link Expr} values from the inputs to this stage.
    *

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

@@ -288,7 +288,10 @@ export class FindNearest implements Stage {
 export class Limit implements Stage {
   name = 'limit';
 
-  constructor(readonly limit: number) {
+  constructor(
+    readonly limit: number,
+    readonly convertedFromLimitTolast: boolean = false
+  ) {
     hardAssert(
       !isNaN(limit) && limit !== Infinity && limit !== -Infinity,
       'Invalid limit value'

packages/firestore/src/local/indexeddb_target_cache.ts

@@ -273,7 +273,6 @@ export class IndexedDbTargetCache implements TargetCache {
           const found = fromDbTarget(this.serializer, value);
           // After finding a potential match, check that the target is
           // actually equal to the requested target.
-          // TODO(pipeline): This needs to handle pipeline properly.
           if (targetOrPipelineEqual(target, found.target)) {
             result = found;
             control.done();

packages/firestore/src/local/local_documents_view.ts

@@ -568,8 +568,6 @@ export class LocalDocumentsView {
     context?: QueryContext
   ): PersistencePromise<DocumentMap> {
     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);
       const collectionId = getPipelineCollectionGroup(pipeline)!;
       let results = documentMap();
       return this.indexManager

packages/firestore/src/local/local_store_impl.ts

@@ -99,11 +99,13 @@ import { Pipeline } from '../lite-api/pipeline';
 
 import {
   canonifyTargetOrPipeline,
+  getPipelineDocuments,
   isPipeline,
   QueryOrPipeline,
   TargetOrPipeline,
   targetOrPipelineEqual
 } from '../core/pipeline-util';
+import { CorePipeline } from '../core/pipeline_run';
 
 export const LOG_TAG = 'LocalStore';
 
@@ -1041,12 +1043,6 @@ export async function localStoreReleaseTarget(
   const localStoreImpl = debugCast(localStore, LocalStoreImpl);
   const targetData = localStoreImpl.targetDataByTarget.get(targetId);
 
-  // TODO(pipeline): this is a hack that only works because pipelines are the only ones returning nulls here.
-  // REMOVE ASAP.
-  if (targetData === null) {
-    return;
-  }
-
   debugAssert(
     targetData !== null,
     `Tried to release nonexistent target: ${targetId}`
@@ -1086,7 +1082,7 @@ export async function localStoreReleaseTarget(
   localStoreImpl.targetDataByTarget =
     localStoreImpl.targetDataByTarget.remove(targetId);
   // TODO(pipeline): This needs to handle pipeline properly.
-  localStoreImpl.targetIdByTarget.delete(targetData!.target as Target);
+  localStoreImpl.targetIdByTarget.delete(targetData!.target);
 }
 
 /**
@@ -1264,6 +1260,24 @@ export function localStoreGetCachedTarget(
   }
 }
 
+// PORTING NOTE: Multi-Tab only.
+export function localStoreGetDocuments(
+  localStore: LocalStore,
+  pipeline: CorePipeline
+): Promise<DocumentMap> {
+  const localStoreImpl = debugCast(localStore, LocalStoreImpl);
+
+  const keys = getPipelineDocuments(pipeline)!;
+  const keySet = documentKeySet(...keys.map(k => DocumentKey.fromPath(k)));
+  return localStoreImpl.persistence
+    .runTransaction('Get documents for pipeline', 'readonly', txn =>
+      localStoreImpl.remoteDocuments.getEntries(txn, keySet)
+    )
+    .then(changedDocs => {
+      return changedDocs;
+    });
+}
+
 /**
  * Returns the set of documents that have been updated since the last call.
  * If this is the first call, returns the set of changes since client