Ability to specify union

JAVA-3472

Author: jstewart148

driver-core/src/main/com/mongodb/client/model/Aggregates.java

@@ -377,6 +377,20 @@ public static <TExpression> Bson group(@Nullable final TExpression id, final Lis
         return new GroupStage<TExpression>(id, fieldAccumulators);
     }
 
+    /**
+     * Creates a $unionWith pipeline stage.
+     *
+     * @param collection    the name of the collection in the same database to perform the union with.
+     * @param pipeline      the pipeline to run on the union.
+     * @return the $unionWith pipeline stage
+     * @mongodb.driver.manual reference/operator/aggregation/unionWith/ $unionWith
+     * @mongodb.server.release 4.4
+     * @since 4.1
+     */
+    public static Bson unionWith(final String collection, final List<? extends Bson> pipeline) {
+        return new UnionWithStage(collection, pipeline);
+    }
+
     /**
      * Creates a $unwind pipeline stage for the specified field name, which must be prefixed by a {@code '$'} sign.
      *
@@ -1376,6 +1390,70 @@ public String toString() {
         }
     }
 
+    private static final class UnionWithStage implements Bson {
+        private final String collection;
+        private final List<? extends Bson> pipeline;
+
+        private UnionWithStage(final String collection, final List<? extends Bson> pipeline) {
+            this.collection = collection;
+            this.pipeline = pipeline;
+        }
+
+        @Override
+        public <TDocument> BsonDocument toBsonDocument(final Class<TDocument> tDocumentClass, final CodecRegistry codecRegistry) {
+            BsonDocumentWriter writer = new BsonDocumentWriter(new BsonDocument());
+
+            writer.writeStartDocument();
+
+            writer.writeStartDocument("$unionWith");
+            writer.writeString("coll", collection);
+
+            writer.writeName("pipeline");
+            writer.writeStartArray();
+            for (Bson stage : pipeline) {
+                BuildersHelper.encodeValue(writer, stage, codecRegistry);
+            }
+            writer.writeEndArray();
+
+            writer.writeEndDocument();
+
+            return writer.getDocument();
+        }
+
+        @Override
+        public boolean equals(final Object o) {
+            if (this == o) {
+                return true;
+            }
+            if (o == null || getClass() != o.getClass()) {
+                return false;
+            }
+
+            UnionWithStage that = (UnionWithStage) o;
+
+            if (!collection.equals(that.collection)) {
+                return false;
+            }
+            return pipeline != null ? !pipeline.equals(that.pipeline) : that.pipeline != null;
+        }
+
+        @Override
+        public int hashCode() {
+            int result = collection.hashCode();
+            result = 31 * result + (pipeline != null ? pipeline.hashCode() : 0);
+            return result;
+        }
+
+        @Override
+        public String toString() {
+            return "Stage{"
+                    + "name='$unionWith'"
+                    + ", collection='" + collection + '\''
+                    + ", pipeline=" + pipeline
+                    + '}';
+        }
+    }
+
     private Aggregates() {
     }
 }

driver-core/src/test/functional/com/mongodb/client/model/AggregatesFunctionalSpecification.groovy

@@ -56,6 +56,7 @@ import static com.mongodb.client.model.Aggregates.sample
 import static com.mongodb.client.model.Aggregates.skip
 import static com.mongodb.client.model.Aggregates.sort
 import static com.mongodb.client.model.Aggregates.sortByCount
+import static com.mongodb.client.model.Aggregates.unionWith
 import static com.mongodb.client.model.Aggregates.unwind
 import static com.mongodb.client.model.Filters.eq
 import static com.mongodb.client.model.Filters.exists
@@ -981,4 +982,41 @@ class AggregatesFunctionalSpecification extends OperationFunctionalSpecification
         then:
         results == [Document.parse('{b: 1, _id: 7}')]
     }
+
+    @IgnoreIf({ !serverVersionAtLeast(4, 3) })
+    def '$unionWith'() {
+        given:
+        def coll1Helper = getCollectionHelper(new MongoNamespace(getDatabaseName(), 'coll1'))
+        def coll2Helper = getCollectionHelper(new MongoNamespace(getDatabaseName(), 'coll2'))
+
+        coll1Helper.drop()
+        coll2Helper.drop()
+
+        coll1Helper.insertDocuments(
+                Document.parse('{ "name1" : "almonds" }'),
+                Document.parse('{ "name1" : "cookies" }'))
+
+        coll2Helper.insertDocuments(
+                Document.parse('{ "name2" : "cookies" }'),
+                Document.parse('{ "name2" : "cookies" }'),
+                Document.parse('{ "name2" : "pecans" }'))
+
+        def pipeline = asList(match(eq('name2', 'cookies')), project(fields(excludeId(), computed('name', '$name2'))),
+                sort(ascending('name')))
+
+        when:
+        def results = coll1Helper.aggregate([project(fields(excludeId(), computed('name', '$name1'))),
+                                             unionWith('coll2', pipeline)])
+
+        then:
+        results == [
+                Document.parse('{ name: "almonds" }'),
+                Document.parse('{ name: "cookies" }'),
+                Document.parse('{ name: "cookies" }'),
+                Document.parse('{ name: "cookies" }') ]
+
+        cleanup:
+        coll1Helper?.drop()
+        coll2Helper?.drop()
+    }
 }

driver-core/src/test/unit/com/mongodb/client/model/AggregatesSpecification.groovy

@@ -56,6 +56,7 @@ import static com.mongodb.client.model.Aggregates.sample
 import static com.mongodb.client.model.Aggregates.skip
 import static com.mongodb.client.model.Aggregates.sort
 import static com.mongodb.client.model.Aggregates.sortByCount
+import static com.mongodb.client.model.Aggregates.unionWith
 import static com.mongodb.client.model.Aggregates.unwind
 import static com.mongodb.client.model.BsonHelper.toBson
 import static com.mongodb.client.model.Filters.eq
@@ -327,6 +328,13 @@ class AggregatesSpecification extends Specification {
         toBson(skip(5)) == parse('{ $skip : 5 }')
     }
 
+    def 'should render $unionWith'() {
+        expect:
+        List<Bson> pipeline = asList(match(expr(new Document('$eq', asList('x', '1')))))
+        toBson(unionWith('with', pipeline)) ==
+                parse('''{ $unionWith : { coll: "with", pipeline : [{ $match : { $expr: { $eq : [ "x" , "1" ]}}}] }}''')
+    }
+
     def 'should render $unwind'() {
         expect:
         toBson(unwind('$sizes')) == parse('{ $unwind : "$sizes" }')

driver-scala/src/main/scala/org/mongodb/scala/model/Aggregates.scala

@@ -431,4 +431,15 @@ object Aggregates {
   def merge(namespace: MongoNamespace, mergeOptions: MergeOptions): Bson =
     JAggregates.merge(namespace, mergeOptions.wrapped)
 
+  /**
+   * Creates a `\$unionWith` pipeline stage.
+   *
+   * @param collection    the name of the collection in the same database to perform the union with.
+   * @param pipeline      the pipeline to run on the union.
+   * @return the \$unionWith` pipeline stage
+   * @see [[http://docs.mongodb.org/manual/reference/operator/aggregation/unionWith/]]
+   */
+  def unionWith(collection: String, pipeline: Bson*): Bson =
+    JAggregates.unionWith(collection, pipeline.asJava)
+
 }