Add batched merge results size estimations to CB

Author: ivancea

server/src/main/java/org/elasticsearch/action/search/QueryPhaseResultConsumer.java

@@ -231,6 +231,7 @@ public SearchPhaseController.ReducedQueryPhase reduce() throws Exception {
         while ((batchedResult = batchedResults.poll()) != null) {
             topDocsStats.add(batchedResult.v1());
             consumePartialMergeResult(batchedResult.v2(), topDocsList, aggsList);
+            addEstimateAndMaybeBreak(batchedResult.v2().estimatedSize);
         }
         for (QuerySearchResult result : buffer) {
             topDocsStats.add(result.topDocs(), result.searchTimedOut(), result.terminatedEarly());
@@ -392,7 +393,7 @@ private MergeResult partialReduce(
         return new MergeResult(
             processedShards,
             newTopDocs,
-            newAggs == null ? null : DelayableWriteable.referencing(newAggs),
+            newAggs != null ? DelayableWriteable.referencing(newAggs) : null,
             newAggs != null ? DelayableWriteable.getSerializedSize(newAggs) : 0
         );
     }