Aggregations: Only return aggregations on the first page when scrolling.

Aggregations are collection-wide statistics so they would always be the same.
In order to save CPU/bandwidth, we can just return them on the first page.

Same as #1642 but for aggregations.

docs/reference/search/request/scroll.asciidoc

@@ -73,6 +73,9 @@ IMPORTANT: The initial search request and each subsequent scroll request
 returns a new `scroll_id` -- only the most recent `scroll_id` should be
 used.
 
+NOTE: If the request specifies aggregations, only the initial search response
+will contain the aggregations results.
+
 [[scroll-scan]]
 ==== Efficient scrolling with Scroll-Scan
 

src/main/java/org/elasticsearch/search/aggregations/AggregationPhase.java

@@ -96,6 +96,7 @@ public class AggregationPhase implements SearchPhase {
     @Override
     public void execute(SearchContext context) throws ElasticsearchException {
         if (context.aggregations() == null) {
+            context.queryResult().aggregations(null);
             return;
         }
 
@@ -133,6 +134,9 @@ public class AggregationPhase implements SearchPhase {
             aggregations.add(aggregator.buildAggregation(0));
         }
         context.queryResult().aggregations(new InternalAggregations(aggregations));
+
+        // disable aggregations so that they don't run on next pages in case of scrolling
+        context.aggregations(null);
     }
 
 

src/test/java/org/elasticsearch/search/aggregations/AggregationsIntegrationTests.java

@@ -0,0 +1,76 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.search.aggregations;
+
+import com.google.common.collect.Lists;
+import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.search.aggregations.bucket.terms.Terms;
+import org.elasticsearch.test.ElasticsearchIntegrationTest;
+
+import java.util.List;
+
+import static org.elasticsearch.search.aggregations.AggregationBuilders.terms;
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertSearchResponse;
+
+
+@ElasticsearchIntegrationTest.SuiteScopeTest
+public class AggregationsIntegrationTests extends ElasticsearchIntegrationTest {
+
+    static int numDocs;
+
+    @Override
+    public void setupSuiteScopeCluster() throws Exception {
+        assertAcked(prepareCreate("index").addMapping("type", "f", "type=string").get());
+        ensureYellow("index");
+        numDocs = randomIntBetween(1, 20);
+        List<IndexRequestBuilder> docs = Lists.newArrayList();
+        for (int i = 0; i < numDocs; ++i) {
+            docs.add(client().prepareIndex("index", "type").setSource("f", Integer.toString(i / 3)));
+        }
+        indexRandom(true, docs);
+    }
+
+    public void testScroll() {
+        final int size = randomIntBetween(1, 4);
+        SearchResponse response = client().prepareSearch("index")
+                .setSize(size).setScroll(new TimeValue(500))
+                .addAggregation(terms("f").field("f")).get();
+        assertSearchResponse(response);
+        Aggregations aggregations = response.getAggregations();
+        assertNotNull(aggregations);
+        Terms terms = aggregations.get("f");
+        assertEquals(Math.min(numDocs, 3L), terms.getBucketByKey("0").getDocCount());
+
+        int total = response.getHits().getHits().length;
+        while (response.getHits().hits().length > 0) {
+            response = client().prepareSearchScroll(response.getScrollId())
+                    .setScroll(new TimeValue(500))
+                    .execute().actionGet();
+            assertNull(response.getAggregations());
+            total += response.getHits().hits().length;
+        }
+        clearScroll(response.getScrollId());
+        assertEquals(numDocs, total);
+    }
+
+}