Add pipeline DML tests and fix update verification

Author: wu-hui

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

@@ -42,12 +42,10 @@
 import com.google.cloud.firestore.pipeline.stages.Aggregate;
 import com.google.cloud.firestore.pipeline.stages.AggregateOptions;
 import com.google.cloud.firestore.pipeline.stages.Delete;
-import com.google.cloud.firestore.pipeline.stages.DeleteOptions;
 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.InsertOptions;
 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,9 +61,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.UpdateOptions;
 import com.google.cloud.firestore.pipeline.stages.Upsert;
-import com.google.cloud.firestore.pipeline.stages.UpsertOptions;
 import com.google.cloud.firestore.pipeline.stages.Where;
 import com.google.cloud.firestore.telemetry.MetricsUtil.MetricsContext;
 import com.google.cloud.firestore.telemetry.TelemetryConstants;
@@ -1013,29 +1009,6 @@ public Pipeline delete() {
     return append(new Delete());
   }
 
-  /**
-   * Performs a delete operation on documents from previous stages.
-   *
-   * @param target The collection to delete from.
-   * @return A new {@code Pipeline} object with this stage appended to the stage list.
-   */
-  @BetaApi
-  public Pipeline delete(CollectionReference target) {
-    return append(Delete.withCollection(target));
-  }
-
-  /**
-   * Performs a delete operation on documents from previous stages.
-   *
-   * @param deleteStage The {@code Delete} stage to append.
-   * @param options The {@code DeleteOptions} to apply to the stage.
-   * @return A new {@code Pipeline} object with this stage appended to the stage list.
-   */
-  @InternalApi
-  public Pipeline delete(Delete deleteStage, DeleteOptions options) {
-    return append(deleteStage.withOptions(options));
-  }
-
   /**
    * Performs an upsert operation using documents from previous stages.
    *
@@ -1054,19 +1027,18 @@ public Pipeline upsert() {
    */
   @BetaApi
   public Pipeline upsert(CollectionReference target) {
-    return append(Upsert.withCollection(target));
+    return append(new Upsert().withCollection(target));
   }
 
   /**
    * Performs an upsert operation using documents from previous stages.
    *
    * @param upsertStage The {@code Upsert} stage to append.
-   * @param options The {@code UpsertOptions} to apply to the stage.
    * @return A new {@code Pipeline} object with this stage appended to the stage list.
    */
   @InternalApi
-  public Pipeline upsert(Upsert upsertStage, UpsertOptions options) {
-    return append(upsertStage.withOptions(options));
+  public Pipeline upsert(Upsert upsertStage) {
+    return append(upsertStage);
   }
 
   /**
@@ -1094,12 +1066,11 @@ public Pipeline update(Selectable... transformations) {
    * Performs an update operation using documents from previous stages.
    *
    * @param update The {@code Update} stage to append.
-   * @param options The {@code UpdateOptions} to apply to the stage.
    * @return A new {@code Pipeline} object with this stage appended to the stage list.
    */
   @InternalApi
-  public Pipeline update(Update updateStage, UpdateOptions options) {
-    return append(updateStage.withOptions(options));
+  public Pipeline update(Update update) {
+    return append(update);
   }
 
   /**
@@ -1120,19 +1091,18 @@ public Pipeline insert() {
    */
   @BetaApi
   public Pipeline insert(CollectionReference target) {
-    return append(Insert.withCollection(target));
+    return append(new Insert().withCollection(target));
   }
 
   /**
    * Performs an insert operation using documents from previous stages.
    *
    * @param insertStage The {@code Insert} stage to append.
-   * @param options The {@code InsertOptions} to apply to the stage.
    * @return A new {@code Pipeline} object with this stage appended to the stage list.
    */
   @InternalApi
-  public Pipeline insert(Insert insertStage, InsertOptions options) {
-    return append(insertStage.withOptions(options));
+  public Pipeline insert(Insert insertStage) {
+    return append(insertStage);
   }
 
   /**
@@ -1476,7 +1446,7 @@ public void onComplete() {
           }
         };
 
-    logger.log(Level.FINEST, "Sending pipeline request: " + request.getStructuredPipeline());
+    logger.log(Level.WARNING, "Sending pipeline request: " + request.getStructuredPipeline());
 
     rpcContext.streamRequest(request, observer, rpcContext.getClient().executePipelineCallable());
   }

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

@@ -24,6 +24,7 @@
 import com.google.cloud.firestore.pipeline.stages.CollectionOptions;
 import com.google.cloud.firestore.pipeline.stages.Database;
 import com.google.cloud.firestore.pipeline.stages.Documents;
+import com.google.cloud.firestore.pipeline.stages.Literals;
 import com.google.common.base.Preconditions;
 import java.util.Arrays;
 import javax.annotation.Nonnull;
@@ -157,6 +158,19 @@ public Pipeline documents(String... docs) {
                 .toArray(DocumentReference[]::new)));
   }
 
+  /**
+   * Creates a new {@link Pipeline} that operates on a static set of documents
+   * represented as Maps.
+   *
+   * @param data The Maps representing documents to include in the pipeline.
+   * @return A new {@code Pipeline} instance with a literals source.
+   */
+  @Nonnull
+  @BetaApi
+  public final Pipeline literals(java.util.Map<String, Object>... data) {
+    return new Pipeline(this.rpcContext, new Literals(data));
+  }
+
   /**
    * Creates a new {@link Pipeline} from the given {@link Query}. Under the hood, this will
    * translate the query semantics (order by document ID, etc.) to an equivalent pipeline.

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

@@ -27,42 +27,17 @@
 
 @InternalApi
 public final class Delete extends Stage {
-
-  @Nullable private final String path;
-
-  private Delete(@Nullable String path, InternalOptions options) {
+  private Delete(InternalOptions options) {
     super("delete", options);
-    this.path = path;
   }
 
   @BetaApi
   public Delete() {
-    this(null, InternalOptions.EMPTY);
-  }
-
-  @BetaApi
-  public static Delete withCollection(CollectionReference target) {
-    String path = target.getPath();
-    return new Delete(path.startsWith("/") ? path : "/" + path, InternalOptions.EMPTY);
-  }
-
-  @InternalApi
-  public Delete withOptions(DeleteOptions options) {
-    return new Delete(path, this.options.adding(options));
-  }
-
-  @InternalApi
-  public Delete withReturns(DeleteReturn returns) {
-    return new Delete(
-        path, this.options.with("returns", PipelineUtils.encodeValue(returns.getValue())));
+    this(InternalOptions.EMPTY);
   }
 
   @Override
   Iterable<Value> toStageArgs() {
-    List<Value> args = new ArrayList<>();
-    if (path != null) {
-      args.add(Value.newBuilder().setReferenceValue(path).build());
-    }
-    return args;
+    return new ArrayList<>();
   }
 }

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

@@ -20,59 +20,41 @@
 import com.google.api.core.InternalApi;
 import com.google.cloud.firestore.CollectionReference;
 import com.google.cloud.firestore.PipelineUtils;
+import com.google.cloud.firestore.pipeline.expressions.Expression;
 import com.google.cloud.firestore.pipeline.expressions.Selectable;
 import com.google.firestore.v1.Value;
 import java.util.ArrayList;
 import java.util.List;
 import javax.annotation.Nullable;
 
-@InternalApi
+@BetaApi
 public final class Insert extends Stage {
 
-  @Nullable private final String path;
-
-  private Insert(@Nullable String path, InternalOptions options) {
+  private Insert(InternalOptions options) {
     super("insert", options);
-    this.path = path;
   }
 
   @BetaApi
   public Insert() {
-    this(null, InternalOptions.EMPTY);
+    this(InternalOptions.EMPTY);
   }
 
   @BetaApi
-  public static Insert withCollection(CollectionReference target) {
+  public Insert withCollection(CollectionReference target) {
     String path = target.getPath();
-    return new Insert(path.startsWith("/") ? path : "/" + path, InternalOptions.EMPTY);
-  }
-
-  @InternalApi
-  public Insert withOptions(InsertOptions options) {
-    return new Insert(path, this.options.adding(options));
-  }
-
-  @InternalApi
-  public Insert withReturns(InsertReturn returns) {
+    String normalizedPath = path.startsWith("/") ? path : "/" + path;
     return new Insert(
-        path, this.options.with("returns", PipelineUtils.encodeValue(returns.getValue())));
+        this.options.with(
+            "collection", Value.newBuilder().setReferenceValue(normalizedPath).build()));
   }
 
   @BetaApi
-  public Insert withTransformations(Selectable... transformations) {
-    return new Insert(
-        path,
-        this.options.with(
-            "transformations",
-            PipelineUtils.encodeValue(PipelineUtils.selectablesToMap(transformations))));
+  public Insert withIdExpression(Expression idExpr) {
+    return new Insert(this.options.with("document_id", PipelineUtils.encodeValue(idExpr)));
   }
 
   @Override
   Iterable<Value> toStageArgs() {
-    List<Value> args = new ArrayList<>();
-    if (path != null) {
-      args.add(Value.newBuilder().setReferenceValue(path).build());
-    }
-    return args;
+    return java.util.Collections.emptyList();
   }
 }