feat(firestore): add DML stages (Insert, Upsert) and atomic execution option to Java SDK pipelines

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;
@@ -1212,7 +1215,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());
   }
@@ -1251,7 +1253,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 +1276,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,11 +1301,22 @@ 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);
   }
 
+  public Pipeline insert(Insert insert) {
+    return append(insert);
+  }
+
+  public Pipeline upsert(Upsert upsert) {
+    return append(upsert);
+  }
+
+  public Pipeline upsert(Selectable... additionalFields) {
+    return append(new Upsert(additionalFields));
+  }
+
   /**
    * Performs an insert operation using documents from previous stages. Adds a generic stage to the
    * pipeline.
@@ -1509,31 +1520,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/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,79 @@
+/*
+ * 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.api.core.InternalApi;
+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;
+
+@InternalApi
+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;
+  }
+
+  public Insert() {
+    this(null, null, InternalOptions.EMPTY);
+  }
+
+  public Insert withCollection(String collectionPath) {
+    return new Insert(collectionPath, this.documentIdExpression, this.options);
+  }
+
+  public Insert withDocumentIdExpression(Expression documentIdExpression) {
+    return new Insert(this.collectionPath, documentIdExpression, this.options);
+  }
+
+  public Insert withDocumentId(Expression documentIdExpression) {
+    return withDocumentIdExpression(documentIdExpression);
+  }
+
+  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<>();
+  }
+}

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

@@ -17,18 +17,24 @@
 package com.google.cloud.firestore.pipeline.stages;
 
 public final class PipelineExecuteOptions extends AbstractOptions<PipelineExecuteOptions> {
+  private final boolean atomic;
 
   public PipelineExecuteOptions() {
-    super(InternalOptions.EMPTY);
+    this(InternalOptions.EMPTY, false);
   }
 
   PipelineExecuteOptions(InternalOptions options) {
+    this(options, false);
+  }
+
+  PipelineExecuteOptions(InternalOptions options, boolean atomic) {
     super(options);
+    this.atomic = atomic;
   }
 
   @Override
   PipelineExecuteOptions self(InternalOptions options) {
-    return new PipelineExecuteOptions(options);
+    return new PipelineExecuteOptions(options, this.atomic);
   }
 
   public PipelineExecuteOptions withExplainOptions(ExplainOptions options) {
@@ -38,4 +44,12 @@ public PipelineExecuteOptions withExplainOptions(ExplainOptions options) {
   public PipelineExecuteOptions withIndexMode(String indexMode) {
     return with("index_mode", indexMode);
   }
+
+  public PipelineExecuteOptions withAtomic(boolean atomic) {
+    return new PipelineExecuteOptions(this.options, atomic);
+  }
+
+  public boolean isAtomic() {
+    return this.atomic;
+  }
 }

java-firestore/google-cloud-firestore/src/main/java/com/google/cloud/firestore/pipeline/stages/Update.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.cloud.firestore.PipelineUtils;
 import com.google.cloud.firestore.pipeline.expressions.Expression;
@@ -38,12 +37,10 @@ private Update(@Nullable Selectable[] transformedFields, InternalOptions options
     this.transformedFields = transformedFields;
   }
 
-  @BetaApi
   public Update() {
     this(null, InternalOptions.EMPTY);
   }
 
-  @BetaApi
   public Update withTransformedFields(Selectable... transformedFields) {
     return new Update(transformedFields, this.options);
   }