feat(firestore): DML

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,7 +1218,6 @@ 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());
   }
@@ -1223,6 +1228,11 @@ public Pipeline delete() {
    * <p>This method updates the documents in place based on the data flowing through the pipeline.
    * To specify transformations, use {@link #update(Selectable...)}.
    *
+   * <p>Each document is written back to its existing path. 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: 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.
    *
    * <pre>{@code
@@ -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 back to its own path, 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);
   }

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

@@ -0,0 +1,126 @@
+/*
+ * Copyright 2026 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.firestore.pipeline.stages;
+
+import com.google.cloud.firestore.PipelineUtils;
+import com.google.cloud.firestore.pipeline.expressions.Expression;
+import com.google.common.collect.ImmutableMap;
+import com.google.firestore.v1.Value;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Map;
+import javax.annotation.Nullable;
+
+/**
+ * A pipeline stage that inserts each document produced by the previous stages as a new document.
+ *
+ * <p>The insert fails with {@code ALREADY_EXISTS} if a target document already exists.
+ *
+ * <p>The target path of each document is determined as follows:
+ *
+ * <ul>
+ *   <li>With neither a collection nor a document ID expression, the input document's own path (its
+ *       {@code __name__} field) is used, so inserting documents read from the database fails with
+ *       {@code ALREADY_EXISTS}, and a document without a {@code __name__} field fails with {@code
+ *       INVALID_ARGUMENT}.
+ *   <li>With only a collection, the document is written to that collection under the input
+ *       document's ID, or under a generated ID if the input document has no {@code __name__}.
+ *   <li>With only a document ID expression, the document is written to the input document's parent
+ *       collection under the ID that the expression evaluates to.
+ *   <li>With both, the document is written to the collection under the ID that the expression
+ *       evaluates to.
+ * </ul>
+ *
+ * <p>Example:
+ *
+ * <pre>{@code
+ * // Copy a document into another collection under a new ID.
+ * firestore.pipeline()
+ *     .collection("books")
+ *     .where(field("title").equal("Dune"))
+ *     .insert(new Insert().withCollection("archive").withDocumentIdExpression(constant("dune")))
+ *     .execute()
+ *     .get();
+ * }</pre>
+ *
+ * @see com.google.cloud.firestore.Pipeline#insert(Insert)
+ */
+public final class Insert extends Stage {
+
+  @Nullable private final String collectionPath;
+  @Nullable private final Expression documentIdExpression;
+
+  private Insert(
+      @Nullable String collectionPath,
+      @Nullable Expression documentIdExpression,
+      InternalOptions options) {
+    super("insert", buildOptions(collectionPath, documentIdExpression, options));
+    this.collectionPath = collectionPath;
+    this.documentIdExpression = documentIdExpression;
+  }
+
+  /**
+   * Creates an insert stage with no target collection and no document ID expression, so each
+   * document is inserted at the input document's own path.
+   */
+  public Insert() {
+    this(null, null, InternalOptions.EMPTY);
+  }
+
+  /**
+   * Sets the collection that documents are inserted into.
+   *
+   * @param collectionPath The path of the target collection, for example {@code "books"} or {@code
+   *     "users/alice/books"}.
+   * @return A new {@code Insert} stage with the target collection set.
+   */
+  public Insert withCollection(String collectionPath) {
+    return new Insert(collectionPath, this.documentIdExpression, this.options);
+  }
+
+  /**
+   * Sets the expression that produces the ID of each inserted document.
+   *
+   * @param documentIdExpression An expression that is evaluated against each input document and
+   *     must evaluate to a document ID (not a document path). Use {@code constant("id")} for a
+   *     fixed ID.
+   * @return A new {@code Insert} stage with the document ID expression set.
+   */
+  public Insert withDocumentIdExpression(Expression documentIdExpression) {
+    return new Insert(this.collectionPath, documentIdExpression, this.options);
+  }
+
+  private static InternalOptions buildOptions(
+      @Nullable String collectionPath,
+      @Nullable Expression documentIdExpression,
+      InternalOptions baseOptions) {
+    Map<String, Value> optsMap = new HashMap<>(baseOptions.options);
+    if (collectionPath != null) {
+      String path = collectionPath.startsWith("/") ? collectionPath : "/" + collectionPath;
+      optsMap.put("collection", Value.newBuilder().setReferenceValue(path).build());
+    }
+    if (documentIdExpression != null) {
+      optsMap.put("document_id", PipelineUtils.encodeValue(documentIdExpression));
+    }
+    return new InternalOptions(ImmutableMap.copyOf(optsMap));
+  }
+
+  @Override
+  Iterable<Value> toStageArgs() {
+    return new ArrayList<>();
+  }
+}