Feature and tests parity with Node SDK

Author: wu-hui

google-cloud-firestore/clirr-ignored-differences.xml

@@ -299,4 +299,21 @@
     <className>com/google/cloud/firestore/collection/StandardComparator</className>
     <to>*</to>
   </difference>
+
+  <!-- Pipeline Queries -->
+  <difference>
+    <differenceType>7012</differenceType>
+    <className>com/google/cloud/firestore/Firestore</className>
+    <method>com.google.cloud.firestore.PipelineSource pipeline()</method>
+  </difference>
+  <difference>
+    <differenceType>7013</differenceType>
+    <className>com/google/cloud/firestore/Transaction</className>
+    <method>com.google.api.core.ApiFuture execute(com.google.cloud.firestore.Pipeline)</method>
+  </difference>
+  <difference>
+    <differenceType>7012</differenceType>
+    <className>com/google/cloud/firestore/spi/v1/FirestoreRpc</className>
+    <method>com.google.api.gax.rpc.ServerStreamingCallable executePipelineCallable()</method>
+  </difference>
 </differences>

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

@@ -16,6 +16,7 @@
 
 package com.google.cloud.firestore;
 
+import static com.google.cloud.firestore.pipeline.expressions.Expr.and;
 import static com.google.cloud.firestore.telemetry.TraceUtil.ATTRIBUTE_KEY_ATTEMPT;
 import static com.google.cloud.firestore.telemetry.TraceUtil.SPAN_NAME_RUN_AGGREGATION_QUERY;
 
@@ -28,7 +29,8 @@
 import com.google.api.gax.rpc.StatusCode;
 import com.google.api.gax.rpc.StreamController;
 import com.google.cloud.Timestamp;
-import com.google.cloud.firestore.pipeline.expressions.ExprWithAlias;
+import com.google.cloud.firestore.pipeline.expressions.AliasedAggregate;
+import com.google.cloud.firestore.pipeline.expressions.BooleanExpr;
 import com.google.cloud.firestore.telemetry.TraceUtil;
 import com.google.cloud.firestore.telemetry.TraceUtil.Scope;
 import com.google.cloud.firestore.v1.FirestoreSettings;
@@ -49,6 +51,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import java.util.stream.Collectors;
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 
@@ -81,12 +84,27 @@ public Query getQuery() {
   @Nonnull
   @BetaApi
   public Pipeline pipeline() {
-    return getQuery()
-        .pipeline()
-        .aggregate(
-            this.aggregateFieldList.stream()
-                .map(PipelineUtils::toPipelineAggregatorTarget)
-                .toArray(ExprWithAlias[]::new));
+    Pipeline pipeline = getQuery().pipeline();
+
+    List<BooleanExpr> existsExprs =
+        this.aggregateFieldList.stream()
+            .map(PipelineUtils::toPipelineExistsExpr)
+            .filter(Objects::nonNull)
+            .collect(Collectors.toList());
+    if (existsExprs.size() == 1) {
+      pipeline = pipeline.where(existsExprs.get(0));
+    } else if (existsExprs.size() > 1) {
+      pipeline =
+          pipeline.where(
+              and(
+                  existsExprs.get(0),
+                  existsExprs.subList(1, existsExprs.size()).toArray(new BooleanExpr[0])));
+    }
+
+    return pipeline.aggregate(
+        this.aggregateFieldList.stream()
+            .map(PipelineUtils::toPipelineAggregatorTarget)
+            .toArray(AliasedAggregate[]::new));
   }
 
   /**