feature: New Pipeline Stages

dev/src/expression.ts

@@ -1942,38 +1942,6 @@ export class Field extends Expr implements Selectable {
   }
 }
 
-/**
- * @beta
- */
-export class Fields extends Expr implements Selectable {
-  exprType: ExprType = 'Field';
-  selectable = true as const;
-
-  private constructor(private fields: Field[]) {
-    super();
-  }
-
-  static of(name: string, ...others: string[]): Fields {
-    return new Fields([Field.of(name), ...others.map(Field.of)]);
-  }
-
-  static ofAll(): Fields {
-    return new Fields([]);
-  }
-
-  fieldList(): Field[] {
-    return this.fields.map(f => f);
-  }
-
-  _toProto(serializer: Serializer): api.IValue {
-    return {
-      arrayValue: {
-        values: this.fields.map(f => f._toProto(serializer)),
-      },
-    };
-  }
-}
-
 /**
  * @beta
  *

dev/src/index.ts

@@ -137,7 +137,6 @@ export {
   Expr,
   ExprWithAlias,
   Field,
-  Fields,
   Constant,
   Function,
   Ordering,

dev/src/pipeline-util.ts

@@ -156,7 +156,6 @@ export class ExecutionUtil<AppModelType> {
         enc,
         callback
       ) => {
-        console.log(`Pipeline response: ${JSON.stringify(proto, null, 2)}`);
         if (proto === NOOP_MESSAGE) {
           callback(undefined);
           return;
@@ -215,14 +214,20 @@ export class ExecutionUtil<AppModelType> {
         // `toProto()` might throw an exception. We rely on the behavior of an
         // async function to convert this exception into the rejected Promise we
         // catch below.
-        const request = pipeline._toProto(
-          transactionOrReadTime,
-          explainOptions
-        );
-
-        console.log(
-          `Executing pipeline: \n ${JSON.stringify(request, null, 2)}`
-        );
+        const request: api.IExecutePipelineRequest = {
+          database: this._firestore.formattedName,
+          structuredPipeline: {
+            pipeline: pipeline._toProto(),
+          },
+        };
+
+        if (transactionOrReadTime instanceof Uint8Array) {
+          request.transaction = transactionOrReadTime;
+        } else if (transactionOrReadTime instanceof Timestamp) {
+          request.readTime = transactionOrReadTime.toProto().timestampValue;
+        } else if (transactionOrReadTime) {
+          request.newTransaction = transactionOrReadTime;
+        }
 
         let streamActive: Deferred<boolean>;
         do {
@@ -392,6 +397,20 @@ export function isFirestoreValue(obj: any): obj is api.IValue {
   return false;
 }
 
+export function selectableToExpr(
+  selectable: firestore.Selectable | string
+): Expr {
+  if (typeof selectable === 'string') {
+    return Field.of(selectable);
+  } else if (selectable instanceof Field) {
+    return selectable;
+  } else if (selectable instanceof ExprWithAlias) {
+    return selectable.expr;
+  } else {
+    throw new Error('unexpected selectable: ' + selectable);
+  }
+}
+
 export function toPipelineFilterCondition(
   f: FilterInternal,
   serializer: Serializer