feat(firestore): DML (#14703)

Co-authored-by: Hui Wu <wuandy@google.com>

Author: dlarocque

java-firestore/google-cloud-firestore/src/main/java/com/google/cloud/firestore/Pipeline.java

@@ -47,6 +47,7 @@
 import com.google.cloud.firestore.pipeline.stages.Distinct;
 import com.google.cloud.firestore.pipeline.stages.FindNearest;
 import com.google.cloud.firestore.pipeline.stages.FindNearestOptions;
+import com.google.cloud.firestore.pipeline.stages.Insert;
 import com.google.cloud.firestore.pipeline.stages.Limit;
 import com.google.cloud.firestore.pipeline.stages.Offset;
 import com.google.cloud.firestore.pipeline.stages.PipelineExecuteOptions;
@@ -63,6 +64,7 @@
 import com.google.cloud.firestore.pipeline.stages.Unnest;
 import com.google.cloud.firestore.pipeline.stages.UnnestOptions;
 import com.google.cloud.firestore.pipeline.stages.Update;
+import com.google.cloud.firestore.pipeline.stages.Upsert;
 import com.google.cloud.firestore.pipeline.stages.Where;
 import com.google.cloud.firestore.telemetry.MetricsUtil.MetricsContext;
 import com.google.cloud.firestore.telemetry.TelemetryConstants;
@@ -76,6 +78,7 @@
 import com.google.firestore.v1.ExecutePipelineRequest;
 import com.google.firestore.v1.ExecutePipelineResponse;
 import com.google.firestore.v1.StructuredPipeline;
+import com.google.firestore.v1.TransactionOptions;
 import com.google.firestore.v1.Value;
 import com.google.protobuf.ByteString;
 import java.util.ArrayList;
@@ -1198,6 +1201,9 @@ public Pipeline unnest(Selectable field, UnnestOptions options) {
   /**
    * Performs a delete operation on documents from previous stages.
    *
+   * <p>Each document produced by the previous stages is deleted from the database. The result
+   * contains a single document with a {@code documents_modified} field.
+   *
    * <p>Example:
    *
    * <pre>{@code
@@ -1212,16 +1218,20 @@ public Pipeline unnest(Selectable field, UnnestOptions options) {
    *
    * @return A new {@code Pipeline} object with this stage appended to the stage list.
    */
-  @BetaApi
   public Pipeline delete() {
     return append(new Delete());
   }
 
   /**
    * Performs an update operation using documents from previous stages.
    *
-   * <p>This method updates the documents in place based on the data flowing through the pipeline.
-   * To specify transformations, use {@link #update(Selectable...)}.
+   * <p>This method updates existing documents based on the data flowing through the pipeline. To
+   * specify transformations, use {@link #update(Selectable...)}.
+   *
+   * <p>Each document is written to the path given by its {@code __name__} field. The stored
+   * document is replaced by the pipeline document, so fields removed by earlier stages (for
+   * example, with {@link #removeFields(String, String...)}) are also removed from the stored
+   * document. This stage does not create documents: target documents that don't exist are skipped.
    *
    * <p>Example 1: Update a collection's schema by adding a new field and removing an old one.
    *
@@ -1235,7 +1245,7 @@ public Pipeline delete() {
    *     .get();
    * }</pre>
    *
-   * <p>Example 2: Update documents in place with data from literals.
+   * <p>Example 2: Update documents with data from literals.
    *
    * <pre>{@code
    * Map<String, Object> updateData = new HashMap<>();
@@ -1251,7 +1261,6 @@ public Pipeline delete() {
    *
    * @return A new {@code Pipeline} object with this stage appended to the stage list.
    */
-  @BetaApi
   public Pipeline update() {
     return append(new Update());
   }
@@ -1275,7 +1284,6 @@ public Pipeline update() {
    * @param transformedFields The transformations to apply.
    * @return A new {@code Pipeline} object with this stage appended to the stage list.
    */
-  @BetaApi
   public Pipeline update(Selectable... transformedFields) {
     return append(new Update().withTransformedFields(transformedFields));
   }
@@ -1301,14 +1309,104 @@ public Pipeline update(Selectable... transformedFields) {
    * @param update The {@code Update} stage to append.
    * @return A new {@code Pipeline} object with this stage appended to the stage list.
    */
-  @BetaApi
   public Pipeline update(Update update) {
     return append(update);
   }
 
   /**
-   * Performs an insert operation using documents from previous stages. Adds a generic stage to the
-   * pipeline.
+   * Inserts each document produced by the previous stages at its own path, given by its {@code
+   * __name__} field.
+   *
+   * <p>This is equivalent to {@code insert(new Insert())}. A document without a {@code __name__}
+   * field fails with {@code INVALID_ARGUMENT}.
+   *
+   * <p>The insert fails with {@code ALREADY_EXISTS} if a target document already exists.
+   *
+   * @return A new {@code Pipeline} object with this stage appended to the stage list.
+   */
+  public Pipeline insert() {
+    return append(new Insert());
+  }
+
+  /**
+   * Inserts each document produced by the previous stages as a new document, using a configured
+   * {@link Insert} stage.
+   *
+   * <p>The insert fails with {@code ALREADY_EXISTS} if a target document already exists. See {@link
+   * Insert} for how the target path of each document is determined.
+   *
+   * <p>Example:
+   *
+   * <pre>{@code
+   * Map<String, Object> book = new HashMap<>();
+   * book.put("title", "Dune");
+   *
+   * firestore.pipeline()
+   *     .literals(book)
+   *     .insert(new Insert().withCollection("books"))
+   *     .execute(new PipelineExecuteOptions().withAtomic(true))
+   *     .get();
+   * }</pre>
+   *
+   * @param insert The {@code Insert} stage to append.
+   * @return A new {@code Pipeline} object with this stage appended to the stage list.
+   */
+  public Pipeline insert(Insert insert) {
+    return append(insert);
+  }
+
+  /**
+   * Writes each document produced by the previous stages, using a configured {@link Upsert} stage.
+   * The target document is created if it doesn't exist and replaced if it does; stored fields are
+   * not merged.
+   *
+   * <p>See {@link Upsert} for how the target path of each document is determined.
+   *
+   * <p>Example:
+   *
+   * <pre>{@code
+   * firestore.pipeline()
+   *     .collection("books")
+   *     .where(field("title").equal("Dune"))
+   *     .upsert(new Upsert().withCollection("archive"))
+   *     .execute()
+   *     .get();
+   * }</pre>
+   *
+   * @param upsert The {@code Upsert} stage to append.
+   * @return A new {@code Pipeline} object with this stage appended to the stage list.
+   */
+  public Pipeline upsert(Upsert upsert) {
+    return append(upsert);
+  }
+
+  /**
+   * Writes each document produced by the previous stages to the path given by its {@code __name__}
+   * field, after adding the given fields. The document is created if it doesn't exist and replaced
+   * if it does; stored fields are not merged. A document without a {@code __name__} field fails
+   * with {@code INVALID_ARGUMENT}.
+   *
+   * <p>Example:
+   *
+   * <pre>{@code
+   * firestore.pipeline()
+   *     .collection("books")
+   *     .where(field("rating").greaterThan(4))
+   *     .upsert(constant("Featured").as("status"))
+   *     .execute()
+   *     .get();
+   * }</pre>
+   *
+   * @param additionalFields Fields to add to each input document before it is written. A field with
+   *     the same name as an existing input field overwrites that field.
+   * @return A new {@code Pipeline} object with this stage appended to the stage list.
+   */
+  public Pipeline upsert(Selectable... additionalFields) {
+    return append(new Upsert(additionalFields));
+  }
+
+  /**
+   * Adds a generic stage to the pipeline.
    *
    * <p>This method provides a flexible way to extend the pipeline's functionality by adding custom
    * stages. Each generic stage is defined by a unique `name` and a set of `params` that control its
@@ -1509,31 +1607,50 @@ public void onError(Throwable t) {
     }
   }
 
-  void executeInternal(
+  ExecutePipelineRequest toExecutePipelineRequest(
       @Nonnull PipelineExecuteOptions options,
       @Nullable final ByteString transactionId,
-      @Nullable com.google.protobuf.Timestamp readTime,
-      PipelineResultObserver observer,
-      MetricsContext metricsContext) {
+      @Nullable com.google.protobuf.Timestamp readTime) {
+    java.util.Map<String, com.google.firestore.v1.Value> optionsMap =
+        new java.util.HashMap<>(StageUtils.toMap(options));
+    optionsMap.remove("atomic");
+
     ExecutePipelineRequest.Builder request =
         ExecutePipelineRequest.newBuilder()
-            .setDatabase(rpcContext.getDatabaseName())
+            .setDatabase(rpcContext != null ? rpcContext.getDatabaseName() : "")
             .setStructuredPipeline(
                 StructuredPipeline.newBuilder()
                     .setPipeline(toProto())
-                    .putAllOptions(StageUtils.toMap(options))
+                    .putAllOptions(optionsMap)
                     .build());
 
     if (transactionId != null) {
       request.setTransaction(transactionId);
+    } else if (options.isAtomic()) {
+      request.setNewTransaction(
+          TransactionOptions.newBuilder()
+              .setReadWrite(TransactionOptions.ReadWrite.getDefaultInstance())
+              .build());
+      request.setAutoCommitTransaction(true);
     }
 
     if (readTime != null) {
       request.setReadTime(readTime);
     }
 
+    return request.build();
+  }
+
+  void executeInternal(
+      @Nonnull PipelineExecuteOptions options,
+      @Nullable final ByteString transactionId,
+      @Nullable com.google.protobuf.Timestamp readTime,
+      PipelineResultObserver observer,
+      MetricsContext metricsContext) {
+    ExecutePipelineRequest request = toExecutePipelineRequest(options, transactionId, readTime);
+
     pipelineInternalStream(
-        request.build(),
+        request,
         new PipelineResultObserver() {
           @Override
           public void onCompleted() {

java-firestore/google-cloud-firestore/src/main/java/com/google/cloud/firestore/PipelineSource.java

@@ -152,6 +152,13 @@ public Pipeline documents(String... docs) {
   /**
    * Creates a new {@link Pipeline} that operates on a static set of documents represented as Maps.
    *
+   * <p>Each map becomes one input document. Field values can be constants or {@link
+   * com.google.cloud.firestore.pipeline.expressions.Expression}s, including expressions nested
+   * inside map or list values. {@link FieldValue} sentinels are not supported; use an expression
+   * such as {@code currentTimestamp()} instead. To give a document a path, set its {@code __name__}
+   * field to a {@link DocumentReference}. Executing a pipeline with no documents fails with {@code
+   * INVALID_ARGUMENT}.
+   *
    * <p>Example:
    *
    * <pre>{@code

java-firestore/google-cloud-firestore/src/main/java/com/google/cloud/firestore/pipeline/stages/Delete.java

@@ -16,7 +16,6 @@
 
 package com.google.cloud.firestore.pipeline.stages;
 
-import com.google.api.core.BetaApi;
 import com.google.api.core.InternalApi;
 import com.google.firestore.v1.Value;
 import java.util.ArrayList;
@@ -27,7 +26,6 @@ private Delete(InternalOptions options) {
     super("delete", options);
   }
 
-  @BetaApi
   public Delete() {
     this(InternalOptions.EMPTY);
   }