Address feedback

Author: wu-hui

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

@@ -32,8 +32,7 @@
 import com.google.api.gax.rpc.StreamController;
 import com.google.api.gax.rpc.UnaryCallable;
 import com.google.cloud.Timestamp;
-import com.google.cloud.firestore.Transaction.AsyncFunction;
-import com.google.cloud.firestore.Transaction.Function;
+import com.google.cloud.firestore.Transaction;
 import com.google.cloud.firestore.spi.v1.FirestoreRpc;
 import com.google.cloud.firestore.telemetry.TraceUtil;
 import com.google.common.annotations.VisibleForTesting;
@@ -411,28 +410,28 @@ public PipelineSource pipeline() {
 
   @Nonnull
   @Override
-  public <T> ApiFuture<T> runTransaction(@Nonnull final Function<T> updateFunction) {
+  public <T> ApiFuture<T> runTransaction(@Nonnull final Transaction.Function<T> updateFunction) {
     return runAsyncTransaction(
         new TransactionAsyncAdapter<>(updateFunction), TransactionOptions.create());
   }
 
   @Nonnull
   @Override
   public <T> ApiFuture<T> runTransaction(
-      @Nonnull final Function<T> updateFunction, @Nonnull TransactionOptions transactionOptions) {
+      @Nonnull final Transaction.Function<T> updateFunction, @Nonnull TransactionOptions transactionOptions) {
     return runAsyncTransaction(new TransactionAsyncAdapter<>(updateFunction), transactionOptions);
   }
 
   @Nonnull
   @Override
-  public <T> ApiFuture<T> runAsyncTransaction(@Nonnull final AsyncFunction<T> updateFunction) {
+  public <T> ApiFuture<T> runAsyncTransaction(@Nonnull final Transaction.AsyncFunction<T> updateFunction) {
     return runAsyncTransaction(updateFunction, TransactionOptions.create());
   }
 
   @Nonnull
   @Override
   public <T> ApiFuture<T> runAsyncTransaction(
-      @Nonnull final AsyncFunction<T> updateFunction,
+      @Nonnull final Transaction.AsyncFunction<T> updateFunction,
       @Nonnull TransactionOptions transactionOptions) {
 
     if (transactionOptions.getReadTime() != null) {
@@ -445,7 +444,7 @@ public <T> ApiFuture<T> runAsyncTransaction(
       // that cannot be tracked client side.
       return new ServerSideTransactionRunner<>(this, updateFunction, transactionOptions).run();
     }
-  }
+}
 
   @Nonnull
   @Override
@@ -543,10 +542,10 @@ public void shutdownNow() {
     closed = true;
   }
 
-  private static class TransactionAsyncAdapter<T> implements AsyncFunction<T> {
-    private final Function<T> syncFunction;
+  private static class TransactionAsyncAdapter<T> implements Transaction.AsyncFunction<T> {
+    private final Transaction.Function<T> syncFunction;
 
-    public TransactionAsyncAdapter(Function<T> syncFunction) {
+    public TransactionAsyncAdapter(Transaction.Function<T> syncFunction) {
       this.syncFunction = syncFunction;
     }
 

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

@@ -813,6 +813,7 @@ private void pipelineInternalStream(
           Timestamp executionTime = null;
           boolean firstResponse = false;
           int numDocuments = 0;
+          int docCounterPerTraceUpdate = 0;
           boolean hasCompleted = false;
 
           @Override
@@ -831,11 +832,14 @@ public void onResponse(ExecutePipelineResponse response) {
             }
             if (response.getResultsCount() > 0) {
               numDocuments += response.getResultsCount();
-              if (numDocuments % 100 == 0) {
+              docCounterPerTraceUpdate += response.getResultsCount();
+              if (numDocuments > 100) {
                 Tracing.getTracer()
                     .getCurrentSpan()
-                    .addAnnotation("Firestore.Query: Received 100 documents");
+                    .addAnnotation("Firestore.Pipeline: Received "+numDocuments+" results");
+                docCounterPerTraceUpdate = 0;
               }
+
               for (Document doc : response.getResultsList()) {
                 resultObserver.onNext(
                     PipelineResult.fromDocument(

google-cloud-firestore/src/main/java/com/google/cloud/firestore/pipeline/PaginatingPipeline.java

@@ -119,7 +119,7 @@ public static <T> Constant of(Iterable<T> value) {
 
   @BetaApi
   public static <T> Constant of(T[] value) {
-    return new Constant(Arrays.asList(value)); // Convert array to list
+    return new Constant(Arrays.asList(value.clone())); // Convert array to list
   }
 
   @BetaApi

google-cloud-firestore/src/main/java/com/google/cloud/firestore/pipeline/expressions/Constant.java

@@ -22,6 +22,7 @@
 import com.google.api.core.InternalApi;
 import com.google.cloud.firestore.DocumentReference;
 import com.google.common.base.Objects;
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Lists;
 import com.google.firestore.v1.Value;
 import java.util.Arrays;
@@ -48,7 +49,7 @@ public class Function implements Expr {
   private final String name;
   private final List<Expr> params;
 
-  protected Function(String name, List<? extends Expr> params) {
+  protected Function(String name, ImmutableList<? extends Expr> params) {
     this.name = name;
     this.params = Collections.unmodifiableList(params);
   }
@@ -2483,7 +2484,7 @@ public static ArrayElement arrayElement() {
    */
   @BetaApi
   public static Function genericFunction(String name, List<Expr> params) {
-    return new Function(name, params);
+    return new Function(name, ImmutableList.copyOf(params));
   }
 
   @Override