Markduckworth/ppl mep

packages/firestore/lite/pipelines/pipelines.ts

@@ -50,11 +50,14 @@ export type {
 
 export { PipelineSource } from '../../src/lite-api/pipeline-source';
 
-export { PipelineResult } from '../../src/lite-api/pipeline-result';
+export {
+  PipelineResult,
+  PipelineSnapshot
+} from '../../src/lite-api/pipeline-result';
 
 export { Pipeline } from '../../src/lite-api/pipeline';
 
-export { pipeline, execute } from '../../src/lite-api/pipeline_impl';
+export { execute } from '../../src/lite-api/pipeline_impl';
 
 export {
   Stage,
@@ -76,6 +79,8 @@ export {
 } from '../../src/lite-api/stage';
 
 export {
+  field,
+  constant,
   add,
   subtract,
   multiply,
@@ -132,79 +137,16 @@ export {
   timestampToUnixSeconds,
   timestampAdd,
   timestampSub,
-  genericFunction,
   ascending,
   descending,
   ExprWithAlias,
   Field,
-  Fields,
   Constant,
-  FirestoreFunction,
-  Add,
-  Subtract,
-  Multiply,
-  Divide,
-  Mod,
-  Eq,
-  Neq,
-  Lt,
-  Lte,
-  Gt,
-  Gte,
-  ArrayConcat,
-  ArrayReverse,
-  ArrayContains,
-  ArrayContainsAll,
-  ArrayContainsAny,
-  ArrayLength,
-  ArrayElement,
-  EqAny,
-  IsNan,
-  Exists,
-  Not,
-  And,
-  Or,
-  Xor,
-  Cond,
-  LogicalMaximum,
-  LogicalMinimum,
-  Reverse,
-  ReplaceFirst,
-  ReplaceAll,
-  CharLength,
-  ByteLength,
-  Like,
-  RegexContains,
-  RegexMatch,
-  StrContains,
-  StartsWith,
-  EndsWith,
-  ToLower,
-  ToUpper,
-  Trim,
-  StrConcat,
-  MapGet,
-  Count,
-  Sum,
-  Avg,
-  Minimum,
-  Maximum,
-  CosineDistance,
-  DotProduct,
-  EuclideanDistance,
-  VectorLength,
-  UnixMicrosToTimestamp,
-  TimestampToUnixMicros,
-  UnixMillisToTimestamp,
-  TimestampToUnixMillis,
-  UnixSecondsToTimestamp,
-  TimestampToUnixSeconds,
-  TimestampAdd,
-  TimestampSub,
+  FunctionExpr,
   Ordering,
   ExprType,
-  AccumulatorTarget,
+  AggregateWithAlias,
   Selectable,
-  FilterCondition,
-  Accumulator
+  BooleanExpr,
+  AggregateFunction
 } from '../../src/lite-api/expressions';

packages/firestore/src/api/pipeline.ts

@@ -15,16 +15,12 @@
  * limitations under the License.
  */
 
-import { firestoreClientExecutePipeline } from '../core/firestore_client';
 import { Pipeline as LitePipeline } from '../lite-api/pipeline';
-import { PipelineResult } from '../lite-api/pipeline-result';
-import { DocumentReference } from '../lite-api/reference';
 import { Stage } from '../lite-api/stage';
 import { UserDataReader } from '../lite-api/user_data_reader';
 import { AbstractUserDataWriter } from '../lite-api/user_data_writer';
-import { cast } from '../util/input_validation';
 
-import { ensureFirestoreConfigured, Firestore } from './database';
+import { Firestore } from './database';
 
 export class Pipeline extends LitePipeline {
   /**
@@ -41,66 +37,8 @@ export class Pipeline extends LitePipeline {
     db: Firestore,
     userDataReader: UserDataReader,
     userDataWriter: AbstractUserDataWriter,
-    stages: Stage[],
-    converter: unknown = {}
+    stages: Stage[]
   ): Pipeline {
     return new Pipeline(db, userDataReader, userDataWriter, stages);
   }
-
-  /**
-   * Executes this pipeline and returns a Promise to represent the asynchronous operation.
-   *
-   * <p>The returned Promise can be used to track the progress of the pipeline execution
-   * and retrieve the results (or handle any errors) asynchronously.
-   *
-   * <p>The pipeline results are returned as a list of {@link PipelineResult} objects. Each {@link
-   * PipelineResult} typically represents a single key/value map that has passed through all the
-   * stages of the pipeline, however this might differ depending on the stages involved in the
-   * pipeline. For example:
-   *
-   * <ul>
-   *   <li>If there are no stages or only transformation stages, each {@link PipelineResult}
-   *       represents a single document.</li>
-   *   <li>If there is an aggregation, only a single {@link PipelineResult} is returned,
-   *       representing the aggregated results over the entire dataset .</li>
-   *   <li>If there is an aggregation stage with grouping, each {@link PipelineResult} represents a
-   *       distinct group and its associated aggregated values.</li>
-   * </ul>
-   *
-   * <p>Example:
-   *
-   * ```typescript
-   * const futureResults = await firestore.pipeline().collection("books")
-   *     .where(gt(Field.of("rating"), 4.5))
-   *     .select("title", "author", "rating")
-   *     .execute();
-   * ```
-   *
-   * @return A Promise representing the asynchronous pipeline execution.
-   */
-  execute(): Promise<PipelineResult[]> {
-    const firestore = cast(this._db, Firestore);
-    const client = ensureFirestoreConfigured(firestore);
-    return firestoreClientExecutePipeline(client, this).then(result => {
-      const docs = result
-        // Currently ignore any response from ExecutePipeline that does
-        // not contain any document data in the `fields` property.
-        .filter(element => !!element.fields)
-        .map(
-          element =>
-            new PipelineResult(
-              this._userDataWriter,
-              element.key?.path
-                ? new DocumentReference(firestore, null, element.key)
-                : undefined,
-              element.fields,
-              element.executionTime?.toTimestamp(),
-              element.createTime?.toTimestamp(),
-              element.updateTime?.toTimestamp()
-            )
-        );
-
-      return docs;
-    });
-  }
 }

packages/firestore/src/api/pipeline_impl.ts

@@ -16,16 +16,16 @@
  */
 
 import { Pipeline } from '../api/pipeline';
-import { toPipeline } from '../core/pipeline-util';
+import { firestoreClientExecutePipeline } from '../core/firestore_client';
 import { Pipeline as LitePipeline } from '../lite-api/pipeline';
-import { PipelineResult } from '../lite-api/pipeline-result';
+import { PipelineResult, PipelineSnapshot } from '../lite-api/pipeline-result';
 import { PipelineSource } from '../lite-api/pipeline-source';
 import { Stage } from '../lite-api/stage';
 import { newUserDataReader } from '../lite-api/user_data_reader';
 import { cast } from '../util/input_validation';
 
-import { Firestore } from './database';
-import { Query } from './reference';
+import { ensureFirestoreConfigured, Firestore } from './database';
+import { DocumentReference } from './reference';
 import { ExpUserDataWriter } from './user_data_writer';
 
 declare module './database' {
@@ -35,49 +35,75 @@ declare module './database' {
 }
 
 /**
- * Experimental Modular API for console testing.
- * @param firestore
- */
-export function pipeline(firestore: Firestore): PipelineSource<Pipeline>;
-
-/**
- * Experimental Modular API for console testing.
- * @param query
+ * Executes this pipeline and returns a Promise to represent the asynchronous operation.
+ *
+ * <p>The returned Promise can be used to track the progress of the pipeline execution
+ * and retrieve the results (or handle any errors) asynchronously.
+ *
+ * <p>The pipeline results are returned as a list of {@link PipelineResult} objects. Each {@link
+ * PipelineResult} typically represents a single key/value map that has passed through all the
+ * stages of the pipeline, however this might differ depending on the stages involved in the
+ * pipeline. For example:
+ *
+ * <ul>
+ *   <li>If there are no stages or only transformation stages, each {@link PipelineResult}
+ *       represents a single document.</li>
+ *   <li>If there is an aggregation, only a single {@link PipelineResult} is returned,
+ *       representing the aggregated results over the entire dataset .</li>
+ *   <li>If there is an aggregation stage with grouping, each {@link PipelineResult} represents a
+ *       distinct group and its associated aggregated values.</li>
+ * </ul>
+ *
+ * <p>Example:
+ *
+ * ```typescript
+ * const futureResults = await execute(firestore.pipeline().collection("books")
+ *     .where(gt(field("rating"), 4.5))
+ *     .select("title", "author", "rating"));
+ * ```
+ *
+ * @param pipeline The pipeline to execute.
+ * @return A Promise representing the asynchronous pipeline execution.
  */
-export function pipeline(query: Query): Pipeline;
+export function execute(pipeline: LitePipeline): Promise<PipelineSnapshot> {
+  const firestore = cast(pipeline._db, Firestore);
+  const client = ensureFirestoreConfigured(firestore);
+  return firestoreClientExecutePipeline(client, pipeline).then(result => {
+    // Get the execution time from the first result.
+    // firestoreClientExecutePipeline returns at least one PipelineStreamElement
+    // even if the returned document set is empty.
+    const executionTime =
+      result.length > 0 ? result[0].executionTime?.toTimestamp() : undefined;
 
-export function pipeline(
-  firestoreOrQuery: Firestore | Query
-): PipelineSource<Pipeline> | Pipeline {
-  if (firestoreOrQuery instanceof Firestore) {
-    const firestore = firestoreOrQuery;
-    return new PipelineSource<Pipeline>((stages: Stage[]) => {
-      return new Pipeline(
-        firestore,
-        newUserDataReader(firestore),
-        new ExpUserDataWriter(firestore),
-        stages
+    const docs = result
+      // Currently ignore any response from ExecutePipeline that does
+      // not contain any document data in the `fields` property.
+      .filter(element => !!element.fields)
+      .map(
+        element =>
+          new PipelineResult(
+            pipeline._userDataWriter,
+            element.key?.path
+              ? new DocumentReference(firestore, null, element.key)
+              : undefined,
+            element.fields,
+            element.createTime?.toTimestamp(),
+            element.updateTime?.toTimestamp()
+          )
       );
-    });
-  } else {
-    const query = firestoreOrQuery;
-    const db = cast<Firestore>(query.firestore, Firestore);
 
-    const litePipeline: LitePipeline = toPipeline(query._query, db);
-    return cast<Pipeline>(litePipeline, Pipeline);
-  }
-}
-
-export function execute(pipeline: LitePipeline): Promise<PipelineResult[]> {
-  return pipeline.execute();
+    return new PipelineSnapshot(pipeline, docs, executionTime);
+  });
 }
 
 // Augment the Firestore class with the pipeline() factory method
 Firestore.prototype.pipeline = function (): PipelineSource<Pipeline> {
-  return pipeline(this);
-};
-
-// Augment the Query class with the pipeline() factory method
-Query.prototype.pipeline = function (): Pipeline {
-  return pipeline(this);
+  return new PipelineSource<Pipeline>(this._databaseId, (stages: Stage[]) => {
+    return new Pipeline(
+      this,
+      newUserDataReader(this),
+      new ExpUserDataWriter(this),
+      stages
+    );
+  });
 };