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

# Conflicts:
#	spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java
#	spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/mapreduce/ReactiveMapReduceTests.java

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

@@ -18,6 +18,7 @@ package org.springframework.data.mongodb.core;
 import static org.springframework.data.mongodb.core.query.Criteria.*;
 import static org.springframework.data.mongodb.core.query.SerializationUtils.*;
 
+import com.mongodb.client.model.MapReduceAction;
 import lombok.AccessLevel;
 import lombok.AllArgsConstructor;
 import lombok.NonNull;
@@ -1727,6 +1728,12 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 			if (mapReduceOptions.getOutputSharded().isPresent()) {
 				result = result.sharded(mapReduceOptions.getOutputSharded().get());
 			}
+
+			MapReduceAction action = mapReduceOptions.getMapReduceAction();
+
+			if(action != null && mapReduceOptions.getOutputCollection() != null){
+				result = result.action(action).collectionName(mapReduceOptions.getOutputCollection());
+			}
 		}
 
 		result = collation.map(Collation::toMongoCollation).map(result::collation).orElse(result);

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);