DATAMONGO-2027 - Consider MapReduce output type.

We now consider the output type (collection output) when rendering the MapReduce command. Previously, all output was returned inline without storing the results in the configured collection.

Original Pull Request: #588

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoTemplate.java

@@ -1809,6 +1809,12 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 			if (mapReduceOptions.getOutputSharded().isPresent()) {
 				mapReduce = mapReduce.sharded(mapReduceOptions.getOutputSharded().get());
 			}
+
+			MapReduceAction action = mapReduceOptions.getMapReduceAction();
+
+			if(action != null && mapReduceOptions.getOutputCollection() != null){
+				mapReduce = mapReduce.action(action).collectionName(mapReduceOptions.getOutputCollection());
+			}
 		}
 
 		mapReduce = collation.map(Collation::toMongoCollation).map(mapReduce::collation).orElse(mapReduce);

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java

@@ -2031,6 +2031,12 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 				publisher = publisher.sharded(options.getOutputSharded().get());
 			}
 
+			MapReduceAction action = options.getMapReduceAction();
+
+			if (action != null && options.getOutputCollection() != null) {
+				publisher = publisher.action(action).collectionName(options.getOutputCollection());
+			}
+
 			publisher = collation.map(Collation::toMongoCollation).map(publisher::collation).orElse(publisher);
 
 			return Flux.from(publisher)

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/mapreduce/MapReduceOptions.java

@@ -24,6 +24,7 @@ import org.springframework.data.mongodb.core.query.Collation;
 import org.springframework.lang.Nullable;
 
 import com.mongodb.MapReduceCommand;
+import com.mongodb.client.model.MapReduceAction;
 
 /**
  * @author Mark Pollack
@@ -295,6 +296,27 @@ public class MapReduceOptions {
 		return collation;
 	}
 
+	/**
+	 * Return the {@link MapReduceAction} derived from {@link com.mongodb.MapReduceCommand.OutputType}.
+	 *
+	 * @return the mapped action or {@literal null} if the action maps to inline output.
+	 * @since 2.0.9
+	 */
+	@Nullable
+	public MapReduceAction getMapReduceAction() {
+
+		switch (outputType) {
+			case MERGE:
+				return MapReduceAction.MERGE;
+			case REDUCE:
+				return MapReduceAction.REDUCE;
+			case REPLACE:
+				return MapReduceAction.REPLACE;
+		}
+
+		return null;
+	}
+
 	public Document getOptionsObject() {
 
 		Document cmd = new Document();
@@ -328,7 +350,7 @@ public class MapReduceOptions {
 
 		Document out = new Document();
 
-		switch (outputType) {
+		switch (getOutputType()) {
 			case INLINE:
 				out.put("inline", 1);
 				break;

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/mapreduce/MapReduceTests.java

@@ -45,6 +45,7 @@ import com.mongodb.client.MongoCollection;
  *
  * @author Mark Pollack
  * @author Thomas Darimont
+ * @author Mark Paluch
  */
 @RunWith(SpringJUnit4ClassRunner.class)
 @ContextConfiguration("classpath:infrastructure.xml")
@@ -73,6 +74,7 @@ public class MapReduceTests {
 		template.dropCollection("jmr1_out");
 		template.dropCollection("jmr1");
 		template.dropCollection("jmrWithGeo");
+		template.dropCollection("mapreduceout");
 	}
 
 	@Test
@@ -240,6 +242,26 @@ public class MapReduceTests {
 		assertEquals(1, m.get("d").intValue());
 	}
 
+	@Test // DATAMONGO-2027
+	public void shouldStoreResultInCollection() {
+
+		createMapReduceData();
+
+		String mapWithExcludeFunction = "function(){ for ( var i=0; i<this.x.length; i++ ){ emit( this.x[i] , 1 ); } }";
+
+		mongoTemplate.mapReduce("jmr1", mapWithExcludeFunction, reduceFunction,
+				new MapReduceOptions().outputCollection("mapreduceout"), ValueObject.class);
+
+		List<ValueObject> results = mongoTemplate.find(new Query(), ValueObject.class, "mapreduceout");
+
+		Map<String, Float> m = copyToMap(results);
+		assertEquals(4, m.size());
+		assertEquals(1, m.get("a").intValue());
+		assertEquals(2, m.get("b").intValue());
+		assertEquals(2, m.get("c").intValue());
+		assertEquals(1, m.get("d").intValue());
+	}
+
 	@Test
 	public void testMapReduceExcludeQuery() {
 		createMapReduceData();
@@ -308,7 +330,7 @@ public class MapReduceTests {
 		c.insertOne(new Document("x", Arrays.asList("c", "d")));
 	}
 
-	private Map<String, Float> copyToMap(MapReduceResults<ValueObject> results) {
+	private Map<String, Float> copyToMap(Iterable<ValueObject> results) {
 		List<ValueObject> valueObjects = new ArrayList<ValueObject>();
 		for (ValueObject valueObject : results) {
 			valueObjects.add(valueObject);

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/mapreduce/ReactiveMapReduceTests.java

@@ -42,6 +42,7 @@ import com.mongodb.reactivestreams.client.Success;
 
 /**
  * @author Christoph Strobl
+ * @author Mark Paluch
  * @currentRead Beyond the Shadows - Brent Weeks
  */
 @RunWith(SpringJUnit4ClassRunner.class)
@@ -60,7 +61,8 @@ public class ReactiveMapReduceTests {
 		StepVerifier
 				.create(template.dropCollection(ValueObject.class) //
 						.mergeWith(template.dropCollection("jmr1")) //
-						.mergeWith(template.dropCollection("jmr1_out"))) //
+						.mergeWith(template.dropCollection("jmr1_out")) //
+						.mergeWith(template.dropCollection("mapreduceout"))) //
 				.verifyComplete();
 	}
 
@@ -79,6 +81,25 @@ public class ReactiveMapReduceTests {
 				.verifyComplete();
 	}
 
+	@Test // DATAMONGO-2027
+	public void shouldStoreResultInCollection() {
+
+		createMapReduceData();
+
+		StepVerifier
+				.create(template.mapReduce(new Query(), Person.class, "jmr1", ValueObject.class, mapFunction, reduceFunction, //
+						MapReduceOptions.options().outputCollection("mapreduceout"))) //
+				.expectNextCount(4) //
+				.verifyComplete();
+
+		StepVerifier.create(template.find(new Query(), ValueObject.class, "mapreduceout").buffer(4)) //
+				.consumeNextWith(result -> {
+					assertThat(result).containsExactlyInAnyOrder(new ValueObject("a", 1), new ValueObject("b", 2),
+							new ValueObject("c", 2), new ValueObject("d", 1));
+				}) //
+				.verifyComplete();
+	}
+
 	@Test // DATAMONGO-1890
 	public void mapReduceWithInlineAndFilterQuery() {