Add Distinct

Author: wu-hui

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

@@ -19,6 +19,7 @@
 import com.google.cloud.firestore.pipeline.stages.Collection;
 import com.google.cloud.firestore.pipeline.stages.CollectionGroup;
 import com.google.cloud.firestore.pipeline.stages.Database;
+import com.google.cloud.firestore.pipeline.stages.Distinct;
 import com.google.cloud.firestore.pipeline.stages.Documents;
 import com.google.cloud.firestore.pipeline.stages.FindNearest;
 import com.google.cloud.firestore.pipeline.stages.GenericStage;
@@ -155,12 +156,12 @@ public Pipeline limit(int limit) {
   }
 
   @BetaApi
-  public Pipeline aggregate(AccumulatorTarget... aggregators) {
+  public Pipeline aggregate(AccumulatorTarget... accumulators) {
     return new Pipeline(
         this.db,
         ImmutableList.<Stage>builder()
             .addAll(stages)
-            .add(Aggregate.newInstance().withAccumulators(aggregators))
+            .add(Aggregate.withAccumulators(accumulators))
             .build());
   }
 
@@ -170,6 +171,26 @@ public Pipeline aggregate(Aggregate aggregate) {
         this.db, ImmutableList.<Stage>builder().addAll(stages).add(aggregate).build());
   }
 
+  @BetaApi
+  public Pipeline distinct(String... fields) {
+    return new Pipeline(
+        this.db,
+        ImmutableList.<Stage>builder()
+            .addAll(stages)
+            .add(new Distinct(PipelineUtils.fieldNamesToMap(fields)))
+            .build());
+  }
+
+  @BetaApi
+  public Pipeline distinct(Selectable... selectables) {
+    return new Pipeline(
+        this.db,
+        ImmutableList.<Stage>builder()
+            .addAll(stages)
+            .add(new Distinct(PipelineUtils.selectablesToMap(selectables)))
+            .build());
+  }
+
   @BetaApi
   public Pipeline findNearest(
       String fieldName, double[] vector, FindNearest.FindNearestOptions options) {

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

@@ -19,11 +19,6 @@ public final class Aggregate implements Stage {
   private final Map<String, Expr> groups;
   private final Map<String, Accumulator> accumulators;
 
-  @BetaApi
-  public static Aggregate newInstance() {
-    return new Aggregate(Collections.emptyMap(), Collections.emptyMap());
-  }
-
   @BetaApi
   public Aggregate withGroups(String... fields) {
     return new Aggregate(PipelineUtils.fieldNamesToMap(fields), this.accumulators);
@@ -35,10 +30,15 @@ public Aggregate withGroups(Selectable... selectables) {
   }
 
   @BetaApi
-  public Aggregate withAccumulators(AccumulatorTarget... aggregators) {
+  public static Aggregate withAccumulators(AccumulatorTarget... accumulators) {
+    if (accumulators.length == 0) {
+      throw new IllegalArgumentException(
+          "Must specify at least one accumulator for aggregate() stage. There is a distinct() stage if only distinct group values are needed.");
+    }
+
     return new Aggregate(
-        this.groups,
-        Arrays.stream(aggregators)
+        Collections.emptyMap(),
+        Arrays.stream(accumulators)
             .collect(
                 Collectors.toMap(
                     AccumulatorTarget::getFieldName, AccumulatorTarget::getAccumulator)));

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

@@ -0,0 +1,28 @@
+package com.google.cloud.firestore.pipeline.stages;
+
+import com.google.api.core.BetaApi;
+import com.google.api.core.InternalApi;
+import com.google.cloud.firestore.pipeline.expressions.Expr;
+import java.util.Map;
+
+@BetaApi
+public final class Distinct implements Stage {
+
+  private static final String name = "distinct";
+  private final Map<String, Expr> groups;
+
+  @InternalApi
+  public Distinct(Map<String, Expr> groups) {
+    this.groups = groups;
+  }
+
+  @InternalApi
+  Map<String, Expr> getGroups() {
+    return groups;
+  }
+
+  @Override
+  public String getName() {
+    return name;
+  }
+}

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

@@ -86,6 +86,12 @@ public static com.google.firestore.v1.Pipeline.Stage toStageProto(Stage stage) {
           .addArgs(encodeValue(aggregateStage.getGroups()))
           .addArgs(encodeValue(aggregateStage.getAccumulators()))
           .build();
+    } else if (stage instanceof Distinct) {
+      Distinct distinctStage = (Distinct) stage;
+      return com.google.firestore.v1.Pipeline.Stage.newBuilder()
+          .setName(distinctStage.getName())
+          .addArgs(encodeValue(distinctStage.getGroups()))
+          .build();
     } else if (stage instanceof FindNearest) {
       FindNearest findNearestStage = (FindNearest) stage;
       return com.google.firestore.v1.Pipeline.Stage.newBuilder()

google-cloud-firestore/src/test/java/com/google/cloud/firestore/it/ITPipelineTest.java

@@ -44,6 +44,7 @@
 import static com.google.cloud.firestore.pipeline.expressions.Function.strConcat;
 import static com.google.cloud.firestore.pipeline.expressions.Function.subtract;
 import static com.google.common.truth.Truth.assertThat;
+import static org.junit.Assert.assertThrows;
 
 import com.google.cloud.firestore.CollectionReference;
 import com.google.cloud.firestore.LocalFirestoreHelper;
@@ -278,16 +279,29 @@ public void testAggregates() throws Exception {
   }
 
   @Test
-  public void testGroupBys() throws Exception {
+  public void testGroupBysWithoutAccumulators() throws Exception {
+    assertThrows(
+        IllegalArgumentException.class,
+        () -> {
+          collection
+              .pipeline()
+              .where(lt("published", 1900))
+              .aggregate(Aggregate.withAccumulators().withGroups("genre"));
+        });
+  }
+
+  @Test
+  public void testDistinct() throws Exception {
     List<PipelineResult> results =
         collection
             .pipeline()
             .where(lt("published", 1900))
-            .aggregate(Aggregate.newInstance().withGroups("genre"))
+            .distinct(Field.of("genre").toLowercase().as("lower_genre"))
             .execute()
             .get();
     assertThat(data(results))
-        .containsExactly(map("genre", "Romance"), map("genre", "Psychological Thriller"));
+        .containsExactly(
+            map("lower_genre", "romance"), map("lower_genre", "psychological thriller"));
   }
 
   @Test
@@ -297,9 +311,7 @@ public void testGroupBysAndAggregate() throws Exception {
             .pipeline()
             .where(lt("published", 1984))
             .aggregate(
-                Aggregate.newInstance()
-                    .withGroups("genre")
-                    .withAccumulators(avg("rating").as("avg_rating")))
+                Aggregate.withAccumulators(avg("rating").as("avg_rating")).withGroups("genre"))
             .where(gt("avg_rating", 4.3))
             .execute()
             .get();
@@ -531,20 +543,21 @@ public void testStrConcat() throws Exception {
   @Test
   public void testStartsWith() throws Exception {
     List<PipelineResult> results =
-        collection.pipeline()
+        collection
+            .pipeline()
             .where(startsWith("title", "The"))
             .select("title")
             .sort(Field.of("title").ascending())
-            .execute().get();
+            .execute()
+            .get();
 
     assertThat(data(results))
         .isEqualTo(
             Lists.newArrayList(
                 map("title", "The Great Gatsby"),
                 map("title", "The Handmaid's Tale"),
                 map("title", "The Hitchhiker's Guide to the Galaxy"),
-                map("title", "The Lord of the Rings")
-                ));
+                map("title", "The Lord of the Rings")));
   }
 
   @Test