Remove assumption that mapReduce always returns statistics

Starting with MongoDB 4.4, MongoDB will no longer produce execution statistics
for the mapReduce command.  The driver no longer assumes that they will be in
the response, and if they are missing will just produce zero values for all
statistics.

JAVA-3522

Author: jyemin

driver-core/src/main/com/mongodb/operation/MapReduceHelper.java

@@ -17,28 +17,29 @@
 package com.mongodb.operation;
 
 import org.bson.BsonDocument;
+import org.bson.BsonInt32;
 
 final class MapReduceHelper {
 
     static MapReduceStatistics createStatistics(final BsonDocument result) {
         return new MapReduceStatistics(getInputCount(result), getOutputCount(result), getEmitCount(result),
-                                       getDuration(result));
+                getDuration(result));
     }
 
     private static int getInputCount(final BsonDocument result) {
-        return result.getDocument("counts").getNumber("input").intValue();
+        return result.getDocument("counts", new BsonDocument()).getNumber("input", new BsonInt32(0)).intValue();
     }
 
     private static int getOutputCount(final BsonDocument result) {
-        return result.getDocument("counts").getNumber("output").intValue();
+        return result.getDocument("counts", new BsonDocument()).getNumber("output", new BsonInt32(0)).intValue();
     }
 
     private static int getEmitCount(final BsonDocument result) {
-        return result.getDocument("counts").getNumber("emit").intValue();
+        return result.getDocument("counts", new BsonDocument()).getNumber("emit", new BsonInt32(0)).intValue();
     }
 
     private static int getDuration(final BsonDocument result) {
-        return result.getNumber("timeMillis").intValue();
+        return result.getNumber("timeMillis", new BsonInt32(0)).intValue();
     }
 
     private MapReduceHelper() {

driver-core/src/test/functional/com/mongodb/operation/MapReduceToCollectionOperationSpecification.groovy

@@ -40,6 +40,8 @@ import spock.lang.IgnoreIf
 import static com.mongodb.ClusterFixture.getBinding
 import static com.mongodb.ClusterFixture.isDiscoverableReplicaSet
 import static com.mongodb.ClusterFixture.serverVersionAtLeast
+import static com.mongodb.ClusterFixture.serverVersionGreaterThan
+import static com.mongodb.ClusterFixture.serverVersionLessThan
 import static com.mongodb.client.model.Filters.gte
 import static java.util.concurrent.TimeUnit.MILLISECONDS
 
@@ -139,6 +141,7 @@ class MapReduceToCollectionOperationSpecification extends OperationFunctionalSpe
         operation.getCollation() == defaultCollation
     }
 
+    @IgnoreIf({ serverVersionGreaterThan('4.2') })
     def 'should return the correct statistics and save the results'() {
         when:
         MapReduceStatistics results = execute(mapReduceOperation, async)
@@ -154,6 +157,22 @@ class MapReduceToCollectionOperationSpecification extends OperationFunctionalSpe
         async << [true, false]
     }
 
+    @IgnoreIf({ serverVersionLessThan('4.3') })
+    def 'should return zero-valued statistics and save the results'() {
+        when:
+        MapReduceStatistics results = execute(mapReduceOperation, async)
+
+        then:
+        results.emitCount == 0
+        results.inputCount == 0
+        results.outputCount == 0
+        helper.count() == 2
+        helper.find() as Set == expectedResults
+
+        where:
+        async << [true, false]
+    }
+
 
     @IgnoreIf({ !serverVersionAtLeast(3, 2) })
     def 'should support bypassDocumentValidation'() {

driver-legacy/src/test/functional/com/mongodb/MapReduceTest.java

@@ -30,6 +30,7 @@
 import static com.mongodb.ClusterFixture.isDiscoverableReplicaSet;
 import static com.mongodb.ClusterFixture.isSharded;
 import static com.mongodb.ClusterFixture.serverVersionAtLeast;
+import static com.mongodb.ClusterFixture.serverVersionLessThan;
 import static com.mongodb.DBObjectMatchers.hasFields;
 import static com.mongodb.DBObjectMatchers.hasSubdocument;
 import static java.util.concurrent.TimeUnit.SECONDS;
@@ -309,11 +310,15 @@ public void shouldReturnStatisticsForInlineMapReduce() {
         MapReduceOutput output = collection.mapReduce(command);
 
         //then
-        //duration is not working on the unstable server version
-        //        assertThat(output.getDuration(), is(greaterThan(0)));
-        assertThat(output.getEmitCount(), is(6));
-        assertThat(output.getInputCount(), is(3));
-        assertThat(output.getOutputCount(), is(4));
+        if (serverVersionLessThan(4, 3)) {
+            assertThat(output.getEmitCount(), is(6));
+            assertThat(output.getInputCount(), is(3));
+            assertThat(output.getOutputCount(), is(4));
+        } else {
+            assertThat(output.getEmitCount(), is(0));
+            assertThat(output.getInputCount(), is(0));
+            assertThat(output.getOutputCount(), is(0));
+        }
     }
 
     @Test
@@ -329,10 +334,17 @@ public void shouldReturnStatisticsForMapReduceIntoACollection() {
         MapReduceOutput output = collection.mapReduce(command);
 
         //then
-        assertThat(output.getDuration(), is(greaterThanOrEqualTo(0)));
-        assertThat(output.getEmitCount(), is(6));
-        assertThat(output.getInputCount(), is(3));
-        assertThat(output.getOutputCount(), is(4));
+        if (serverVersionLessThan(4, 3)) {
+            assertThat(output.getDuration(), is(greaterThanOrEqualTo(0)));
+            assertThat(output.getEmitCount(), is(6));
+            assertThat(output.getInputCount(), is(3));
+            assertThat(output.getOutputCount(), is(4));
+        } else {
+            assertThat(output.getDuration(), is(0));
+            assertThat(output.getEmitCount(), is(0));
+            assertThat(output.getInputCount(), is(0));
+            assertThat(output.getOutputCount(), is(0));
+        }
     }