import org.elasticsearch.search.aggregations.bucket.terms.InternalTerms; //导入依赖的package包/类
private static AggregationResult reduceAggregationResults(ElasticsearchSearchQueryBase query, List<Aggregation> aggs) {
    if (aggs.size() == 0) {
        throw new MemgraphException("Cannot reduce zero sized aggregation list");
    }
    Aggregation first = aggs.get(0);
    if (first instanceof HistogramAggregation || first instanceof InternalHistogram || first instanceof InternalDateHistogram) {
        return reduceHistogramResults(query, aggs);
    }
    if (first instanceof RangeAggregation || first instanceof InternalRange) {
        return reduceRangeResults(query, aggs);
    }
    if (first instanceof PercentilesAggregation || first instanceof Percentiles) {
        return reducePercentilesResults(query, aggs);
    }
    if (first instanceof TermsAggregation || first instanceof InternalTerms) {
        return reduceTermsResults(query, aggs);
    }
    if (first instanceof GeohashAggregation || first instanceof InternalGeoHashGrid) {
        return reduceGeohashResults(query, aggs);
    }
    if (first instanceof StatisticsAggregation || first instanceof InternalExtendedStats) {
        return reduceStatisticsResults(aggs);
    }
    throw new MemgraphException("Unhandled aggregation type: " + first.getClass().getName());
}
 

import org.elasticsearch.search.aggregations.bucket.terms.InternalTerms; //导入依赖的package包/类
private static AggregationResult reduceAggregationResults(ElasticsearchSearchQueryBase query, List<Aggregation> aggs) {
    if (aggs.size() == 0) {
        throw new VertexiumException("Cannot reduce zero sized aggregation list");
    }
    Aggregation first = aggs.get(0);
    if (first instanceof HistogramAggregation || first instanceof InternalHistogram || first instanceof InternalDateHistogram) {
        return reduceHistogramResults(query, aggs);
    }
    if (first instanceof RangeAggregation || first instanceof InternalRange) {
        return reduceRangeResults(query, aggs);
    }
    if (first instanceof PercentilesAggregation || first instanceof Percentiles) {
        return reducePercentilesResults(query, aggs);
    }
    if (first instanceof TermsAggregation || first instanceof InternalTerms) {
        return reduceTermsResults(query, aggs);
    }
    if (first instanceof GeohashAggregation || first instanceof InternalGeoHashGrid) {
        return reduceGeohashResults(query, aggs);
    }
    if (first instanceof StatisticsAggregation || first instanceof InternalExtendedStats) {
        return reduceStatisticsResults(aggs);
    }
    throw new VertexiumException("Unhandled aggregation type: " + first.getClass().getName());
}
 

import org.elasticsearch.search.aggregations.bucket.terms.InternalTerms; //导入依赖的package包/类
public SearchResult(SearchResponse resp, Select select) throws SqlParseException {
	Aggregations aggs = resp.getAggregations();
	if (aggs.get("filter") != null) {
		InternalFilter inf = aggs.get("filter");
		aggs = inf.getAggregations();
	}
	if (aggs.get("group by") != null) {
		InternalTerms terms = aggs.get("group by");
		Collection<Bucket> buckets = terms.getBuckets();
		this.total = buckets.size();
		results = new ArrayList<>(buckets.size());
		for (Bucket bucket : buckets) {
			Map<String, Object> aggsMap = toAggsMap(bucket.getAggregations().getAsMap());
			aggsMap.put("docCount", bucket.getDocCount());
			results.add(aggsMap);
		}
	} else {
		results = new ArrayList<>(1);
		this.total = 1;
		Map<String, Object> map = new HashMap<>();
		for (Aggregation aggregation : aggs) {
			map.put(aggregation.getName(), covenValue(aggregation));
		}
		results.add(map);
	}

}
 

import org.elasticsearch.search.aggregations.bucket.terms.InternalTerms; //导入依赖的package包/类
private SearchResponse nestedAggregationWithTimeout(
        SearchRequestBuilder searchRequestBuilder, int expectedResultSize) {
    int timeout = 10;
    while (timeout > 0) {
        try {
            SearchResponse response = searchRequestBuilder.execute()
                    .actionGet();
            if (response.getFailedShards() == 0) {
                for (Aggregation aggregation : response.getAggregations()) {
                    for (Bucket bucket : ((InternalTerms) aggregation)
                            .getBuckets()) {
                        for (Aggregation nestedAggregation : bucket
                                .getAggregations()) {
                            if (((InternalTerms) nestedAggregation)
                                    .getBuckets().size() >= expectedResultSize) {
                                return response;
                            }
                        }
                    }
                }
                return response;
            }
        } catch (Exception e) {
        }
        try {
            Thread.sleep(1000);
        } catch (Exception ex) {
        }
        timeout--;
    }
    return null;
}
 

import org.elasticsearch.search.aggregations.bucket.terms.InternalTerms; //导入依赖的package包/类
@Test
public void testReportsWithBulkRequestLimit() {
    DecimalFormat formatter = new DecimalFormat("00");
    String metricNamePrefix = "com.codahale.metrics.elasticsearch.test.counter";
    for (int i = 0; i < 15; i++) {
        Counter counter = registry.counter(metricNamePrefix
                + formatter.format(i));
        counter.inc();
    }
    waitForReporter();
    flushAndRefresh();

    SearchResponse searchResponse = nestedAggregationWithTimeout(
            new SearchRequestBuilder(ElasticsearchIntegrationTest.client())
                    .setIndices("_all")
                    .setTypes(MetricElasticsearchTypes.COUNTER)
                    .setQuery(QueryBuilders.matchAllQuery())
                    .setSize(0)
                    .addAggregation(
                            AggregationBuilders
                                    .terms("timestamps")
                                    .size(50)
                                    .field("@timestamp")
                                    .order(Order.term(true))
                                    .subAggregation(
                                            AggregationBuilders
                                                    .terms("names")
                                                    .size(50)
                                                    .field("@name")
                                                    .order(Order.term(true)))),
            15);

    InternalTerms timestamps = searchResponse.getAggregations().get(
            "timestamps");
    Collection<Bucket> timestampBuckets = timestamps.getBuckets();
    Assert.assertEquals(true, timestampBuckets.size() > 0);
    for (Bucket bucket : timestampBuckets) {
        InternalTerms names = bucket.getAggregations().get("names");
        if(names.getBuckets().size() >= 15) {
            for (int i = 0; i < names.getBuckets().size(); i++) {
                Assert.assertEquals(
                        prefix + "." + metricNamePrefix + formatter.format(i),
                        ((Bucket) names.getBuckets().toArray()[i]).getKey());
            }
            return;
        }
    }
    Assert.fail("Insufficient metrics reported");
}