Each document produced by the previous stages is deleted from the database. The result
+ * contains a single document with a {@code documents_modified} field.
+ *
*
Example:
*
*
{@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() {
*
This method updates the documents in place based on the data flowing through the pipeline.
* To specify transformations, use {@link #update(Selectable...)}.
*
+ *
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.
+ *
*
Example 1: Update a collection's schema by adding a new field and removing an old one.
*
*
{@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.
+ *
+ *
This is equivalent to {@code insert(new Insert())}. A document without a {@code __name__}
+ * field fails with {@code INVALID_ARGUMENT}.
+ *
+ *
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.
+ *
+ *
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.
+ *
+ *
+ *
+ * @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.
+ *
+ *
See {@link Upsert} for how the target path of each document is determined.
+ *
+ *
+ *
+ * @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}.
+ *
+ *
+ *
+ * @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.
*
*
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 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() {
diff --git a/java-firestore/google-cloud-firestore/src/main/java/com/google/cloud/firestore/PipelineSource.java b/java-firestore/google-cloud-firestore/src/main/java/com/google/cloud/firestore/PipelineSource.java
index f5c11e76d748..080e10d19a5c 100644
--- a/java-firestore/google-cloud-firestore/src/main/java/com/google/cloud/firestore/PipelineSource.java
+++ b/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.
*
+ *
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}.
+ *
*
Example:
*
*
{@code
diff --git a/java-firestore/google-cloud-firestore/src/main/java/com/google/cloud/firestore/pipeline/stages/Delete.java b/java-firestore/google-cloud-firestore/src/main/java/com/google/cloud/firestore/pipeline/stages/Delete.java
index c288f0d42a3b..b31d2ee35493 100644
--- a/java-firestore/google-cloud-firestore/src/main/java/com/google/cloud/firestore/pipeline/stages/Delete.java
+++ b/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);
}
diff --git a/java-firestore/google-cloud-firestore/src/main/java/com/google/cloud/firestore/pipeline/stages/Insert.java b/java-firestore/google-cloud-firestore/src/main/java/com/google/cloud/firestore/pipeline/stages/Insert.java
new file mode 100644
index 000000000000..c87fab63e16f
--- /dev/null
+++ b/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.
+ *
+ *
The insert fails with {@code ALREADY_EXISTS} if a target document already exists.
+ *
+ *
The target path of each document is determined as follows:
+ *
+ *
+ *
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}.
+ *
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__}.
+ *
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.
+ *
With both, the document is written to the collection under the ID that the expression
+ * evaluates to.
+ *
+ *
+ *
Example:
+ *
+ *
{@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();
+ * }
+ *
+ * @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 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 toStageArgs() {
+ return new ArrayList<>();
+ }
+}
diff --git a/java-firestore/google-cloud-firestore/src/main/java/com/google/cloud/firestore/pipeline/stages/Literals.java b/java-firestore/google-cloud-firestore/src/main/java/com/google/cloud/firestore/pipeline/stages/Literals.java
index f27e172701c7..a3e18cd51805 100644
--- a/java-firestore/google-cloud-firestore/src/main/java/com/google/cloud/firestore/pipeline/stages/Literals.java
+++ b/java-firestore/google-cloud-firestore/src/main/java/com/google/cloud/firestore/pipeline/stages/Literals.java
@@ -18,9 +18,11 @@
import com.google.api.core.InternalApi;
import com.google.cloud.firestore.PipelineUtils;
+import com.google.cloud.firestore.pipeline.expressions.Expression;
import com.google.firestore.v1.Value;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -49,11 +51,13 @@ private Value encodeLiteralMap(Map map) {
for (Map.Entry entry : map.entrySet()) {
String key = String.valueOf(entry.getKey());
Object v = entry.getValue();
- if (v instanceof com.google.cloud.firestore.pipeline.expressions.Expression) {
- mapValue.putFields(
- key,
- PipelineUtils.encodeValue(
- (com.google.cloud.firestore.pipeline.expressions.Expression) v));
+ if (v instanceof Expression) {
+ mapValue.putFields(key, PipelineUtils.encodeValue((Expression) v));
+ } else if (containsExpression(v)) {
+ // The backend evaluates expressions only in the top-level fields of a literal document,
+ // so a map or list that contains a nested expression is sent as a map(...) or array(...)
+ // expression instead of a raw map or array value.
+ mapValue.putFields(key, PipelineUtils.encodeValue(toExpression(v)));
} else if (v instanceof Map) {
mapValue.putFields(key, encodeLiteralMap((Map) v));
} else {
@@ -62,4 +66,48 @@ private Value encodeLiteralMap(Map map) {
}
return Value.newBuilder().setMapValue(mapValue.build()).build();
}
+
+ /** Returns whether {@code value} is a map or list that contains an expression at any depth. */
+ private static boolean containsExpression(Object value) {
+ if (value instanceof Map) {
+ for (Object child : ((Map) value).values()) {
+ if (child instanceof Expression || containsExpression(child)) {
+ return true;
+ }
+ }
+ } else if (value instanceof List) {
+ for (Object child : (List) value) {
+ if (child instanceof Expression || containsExpression(child)) {
+ return true;
+ }
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Converts a map or list that contains nested expressions into an equivalent {@code map(...)} or
+ * {@code array(...)} expression. Children without expressions are kept as constants.
+ */
+ private static Expression toExpression(Object value) {
+ if (value instanceof Expression) {
+ return (Expression) value;
+ } else if (value instanceof Map) {
+ Map elements = new LinkedHashMap<>();
+ for (Map.Entry entry : ((Map) value).entrySet()) {
+ elements.put(String.valueOf(entry.getKey()), toExpressionIfNeeded(entry.getValue()));
+ }
+ return Expression.map(elements);
+ } else {
+ List