feat: DML prototype

dev/src/pipelines/pipelines.ts

@@ -81,6 +81,9 @@ import {
   Sample,
   Union,
   Unnest,
+  DeleteStage,
+  UpsertStage,
+  InsertStage,
   InternalWhereStageOptions,
   InternalOffsetStageOptions,
   InternalLimitStageOptions,
@@ -1500,6 +1503,138 @@ export class Pipeline implements firestore.Pipelines.Pipeline {
     return this._addStage(new Sort(internalOptions));
   }
 
+  /**
+   * @beta
+   * Performs a delete operation on documents from previous stages.
+   *
+   * @return A new {@code Pipeline} object with this stage appended to the stage list.
+   */
+  delete(): Pipeline;
+  /**
+   * @beta
+   * Performs a delete operation on documents from previous stages.
+   *
+   * @param collectionNameOrRef - The collection to delete from.
+   * @return A new {@code Pipeline} object with this stage appended to the stage list.
+   */
+  delete(collectionNameOrRef: string | firestore.CollectionReference): Pipeline;
+  /**
+   * @beta
+   * Performs a delete operation on documents from previous stages.
+   *
+   * @param options - The {@code DeleteStageOptions} to apply to the stage.
+   * @return A new {@code Pipeline} object with this stage appended to the stage list.
+   */
+  delete(options: firestore.Pipelines.DeleteStageOptions): Pipeline;
+  delete(
+    optionsOrCollection?:
+      | string
+      | firestore.CollectionReference
+      | firestore.Pipelines.DeleteStageOptions,
+  ): Pipeline {
+    let target = undefined;
+    if (typeof optionsOrCollection === 'string') {
+      target = this.db.collection(optionsOrCollection);
+    } else if (isCollectionReference(optionsOrCollection)) {
+      target = optionsOrCollection;
+    }
+    const options = (
+      !isCollectionReference(optionsOrCollection) &&
+      typeof optionsOrCollection !== 'string'
+        ? optionsOrCollection
+        : undefined
+    ) as firestore.Pipelines.DeleteStageOptions | undefined;
+    return this._addStage(new DeleteStage(target, options));
+  }
+
+  /**
+   * @beta
+   * Performs an upsert operation using documents from previous stages.
+   *
+   * @return A new {@code Pipeline} object with this stage appended to the stage list.
+   */
+  upsert(): Pipeline;
+  /**
+   * @beta
+   * Performs an upsert operation using documents from previous stages.
+   *
+   * @param collectionNameOrRef - The collection to upsert to.
+   * @return A new {@code Pipeline} object with this stage appended to the stage list.
+   */
+  upsert(collectionNameOrRef: string | firestore.CollectionReference): Pipeline;
+  /**
+   * @beta
+   * Performs an upsert operation using documents from previous stages.
+   *
+   * @param options - The {@code UpsertStageOptions} to apply to the stage.
+   * @return A new {@code Pipeline} object with this stage appended to the stage list.
+   */
+  upsert(options: firestore.Pipelines.UpsertStageOptions): Pipeline;
+  upsert(
+    optionsOrCollection?:
+      | string
+      | firestore.CollectionReference
+      | firestore.Pipelines.UpsertStageOptions,
+  ): Pipeline {
+    let target = undefined;
+    if (typeof optionsOrCollection === 'string') {
+      target = this.db.collection(optionsOrCollection);
+    } else if (isCollectionReference(optionsOrCollection)) {
+      target = optionsOrCollection;
+    }
+    const options = (
+      !isCollectionReference(optionsOrCollection) &&
+      typeof optionsOrCollection !== 'string'
+        ? optionsOrCollection
+        : undefined
+    ) as firestore.Pipelines.UpsertStageOptions | undefined;
+    return this._addStage(new UpsertStage(target, options));
+  }
+
+  /**
+   * @beta
+   * Performs an insert operation using documents from previous stages.
+   *
+   * @return A new {@code Pipeline} object with this stage appended to the stage list.
+   */
+  insert(): Pipeline;
+  /**
+   * @beta
+   * Performs an insert operation using documents from previous stages.
+   *
+   * @param collectionNameOrRef - The collection to insert to.
+   * @return A new {@code Pipeline} object with this stage appended to the stage list.
+   */
+  insert(collectionNameOrRef: string | firestore.CollectionReference): Pipeline;
+  /**
+   * @beta
+   * Performs an insert operation using documents from previous stages.
+   *
+   * @param options - The {@code InsertStageOptions} to apply to the stage.
+   * @return A new {@code Pipeline} object with this stage appended to the stage list.
+   */
+  insert(options: firestore.Pipelines.InsertStageOptions): Pipeline;
+  insert(
+    optionsOrCollection?:
+      | string
+      | firestore.CollectionReference
+      | firestore.Pipelines.InsertStageOptions,
+  ): Pipeline {
+    let target = undefined;
+    if (typeof optionsOrCollection === 'string') {
+      target = this.db.collection(optionsOrCollection);
+    } else if (isCollectionReference(optionsOrCollection)) {
+      target = optionsOrCollection;
+    }
+    const options = (
+      !isCollectionReference(optionsOrCollection) &&
+      typeof optionsOrCollection !== 'string'
+        ? optionsOrCollection
+        : undefined
+    ) as firestore.Pipelines.InsertStageOptions | undefined;
+    return this._addStage(new InsertStage(target, options));
+  }
+
   /**
    * @beta
    * Adds a raw stage to the pipeline.

dev/src/pipelines/stage.ts

@@ -677,3 +677,96 @@ export class RawStage implements Stage {
     };
   }
 }
+
+export class DeleteStage implements Stage {
+  name = 'delete';
+  readonly optionsUtil = new OptionsUtil({
+    returns: {serverName: 'returns'},
+    transactional: {serverName: 'transactional'},
+  });
+
+  constructor(
+    private target?: firestore.CollectionReference,
+    private rawOptions?: firestore.Pipelines.DeleteStageOptions,
+  ) {}
+
+  _toProto(serializer: Serializer): api.Pipeline.IStage {
+    const args: api.IValue[] = [];
+    if (this.target) {
+      args.push({referenceValue: this.target.path});
+    }
+
+    return {
+      name: this.name,
+      args,
+      options: this.optionsUtil.getOptionsProto(
+        serializer,
+        {},
+        this.rawOptions,
+      ),
+    };
+  }
+}
+
+export class UpsertStage implements Stage {
+  name = 'upsert';
+  readonly optionsUtil = new OptionsUtil({
+    returns: {serverName: 'returns'},
+    conflict_resolution: {serverName: 'conflict_resolution'},
+    transformations: {serverName: 'transformations'},
+    transactional: {serverName: 'transactional'},
+  });
+
+  constructor(
+    private target?: firestore.CollectionReference,
+    private rawOptions?: firestore.Pipelines.UpsertStageOptions,
+  ) {}
+
+  _toProto(serializer: Serializer): api.Pipeline.IStage {
+    const args: api.IValue[] = [];
+    if (this.target) {
+      args.push({referenceValue: this.target.path});
+    }
+
+    return {
+      name: this.name,
+      args,
+      options: this.optionsUtil.getOptionsProto(
+        serializer,
+        {},
+        this.rawOptions,
+      ),
+    };
+  }
+}
+
+export class InsertStage implements Stage {
+  name = 'insert';
+  readonly optionsUtil = new OptionsUtil({
+    returns: {serverName: 'returns'},
+    transformations: {serverName: 'transformations'},
+    transactional: {serverName: 'transactional'},
+  });
+
+  constructor(
+    private target?: firestore.CollectionReference,
+    private rawOptions?: firestore.Pipelines.InsertStageOptions,
+  ) {}
+
+  _toProto(serializer: Serializer): api.Pipeline.IStage {
+    const args: api.IValue[] = [];
+    if (this.target) {
+      args.push({referenceValue: this.target.path});
+    }
+
+    return {
+      name: this.name,
+      args,
+      options: this.optionsUtil.getOptionsProto(
+        serializer,
+        {},
+        this.rawOptions,
+      ),
+    };
+  }
+}

dev/system-test/pipeline.ts

@@ -309,6 +309,59 @@
   afterEach(() => verifyInstance(firestore as unknown as InternalFirestore));
 
   describe('pipeline results', () => {
+    describe('DML stages', () => {
+      it('can execute delete stage', async () => {
+        const docRef = randomCol.doc('testDelete');
+        await docRef.set({foo: 'bar'});
+
+        const ppl = firestore
+          .pipeline()
+          .collection(randomCol.path)
+          .where(equal(field('__name__'), 'testDelete'))
+          .delete();
+
+        const res = await ppl.execute();
+        expect(res.results.length).to.equal(0);
+
+        // verify document was deleted
+        const docSnap = await docRef.get();
+        expect(docSnap.exists).to.be.false;
+      });
+
+      it('can execute upsert stage', async () => {
+        const docRef = randomCol.doc('testUpsert');
+
+        const ppl = firestore
+          .pipeline()
+          .collection(randomCol.path)
+          .where(equal(field('__name__'), 'testDelete'))
+          .addFields(
+            // Use selectables for addFields
+            field('__name__').as('id'),
+            'upserted_value' as unknown as Pipelines.Selectable, // Hardcoded values inside addFields need specific treatment or aren't supported
+          )
+          .upsert(randomCol.path);
+
+        await ppl.execute();
+      });
+
+      it('can execute insert stage', async () => {
+        const docRef = randomCol.doc('testInsert');
+
+        const ppl = firestore
+          .pipeline()
+          .collection(randomCol.path)
+          .where(equal(field('__name__'), 'testDelete')) // arbitrary matching
+          .addFields(
+            field('__name__').as('id'),
+            'inserted_value' as unknown as Pipelines.Selectable,
+          )
+          .insert(randomCol.path);
+
+        await ppl.execute();
+      });
+    });
+
     it('empty snapshot as expected', async () => {
       const snapshot = await firestore
         .pipeline()