import org.elasticsearch.search.aggregations.InternalAggregations; //导入依赖的package包/类
public PercolateShardResponse(BytesRef[] matches, List<Map<String, HighlightField>> hls, long count, float[] scores, PercolateContext context, ShardId shardId) {
    super(shardId);
    this.matches = matches;
    this.hls = hls;
    this.count = count;
    this.scores = scores;
    this.percolatorTypeId = context.percolatorTypeId;
    this.requestedSize = context.size();
    QuerySearchResult result = context.queryResult();
    if (result != null) {
        if (result.aggregations() != null) {
            this.aggregations = (InternalAggregations) result.aggregations();
        }
        this.pipelineAggregators = result.pipelineAggregators();
    }
}
 

import org.elasticsearch.search.aggregations.InternalAggregations; //导入依赖的package包/类
@Override
public InternalAggregation reduce(InternalAggregation aggregation, ReduceContext reduceContext) {
    MultiBucketsAggregation histo = (MultiBucketsAggregation) aggregation;
    List<? extends Bucket> buckets = histo.getBuckets();
    HistogramFactory factory = (HistogramFactory) histo;

    List<Bucket> newBuckets = new ArrayList<>();
    double sum = 0;
    for (Bucket bucket : buckets) {
        Double thisBucketValue = resolveBucketValue(histo, bucket, bucketsPaths()[0], GapPolicy.INSERT_ZEROS);
        sum += thisBucketValue;
        List<InternalAggregation> aggs = StreamSupport.stream(bucket.getAggregations().spliterator(), false).map((p) -> {
            return (InternalAggregation) p;
        }).collect(Collectors.toList());
        aggs.add(new InternalSimpleValue(name(), sum, formatter, new ArrayList<PipelineAggregator>(), metaData()));
        Bucket newBucket = factory.createBucket(factory.getKey(bucket), bucket.getDocCount(), new InternalAggregations(aggs));
        newBuckets.add(newBucket);
    }
    return factory.createAggregation(newBuckets);
}
 

import org.elasticsearch.search.aggregations.InternalAggregations; //导入依赖的package包/类
ReducedQueryPhase(long totalHits, long fetchHits, float maxScore, boolean timedOut, Boolean terminatedEarly,
                         QuerySearchResult oneResult, Suggest suggest, InternalAggregations aggregations,
                         SearchProfileShardResults shardResults, int numReducePhases) {
    if (numReducePhases <= 0) {
        throw new IllegalArgumentException("at least one reduce phase must have been applied but was: " + numReducePhases);
    }
    this.totalHits = totalHits;
    this.fetchHits = fetchHits;
    if (Float.isInfinite(maxScore)) {
        this.maxScore = Float.NaN;
    } else {
        this.maxScore = maxScore;
    }
    this.timedOut = timedOut;
    this.terminatedEarly = terminatedEarly;
    this.oneResult = oneResult;
    this.suggest = suggest;
    this.aggregations = aggregations;
    this.shardResults = shardResults;
    this.numReducePhases = numReducePhases;
}
 

import org.elasticsearch.search.aggregations.InternalAggregations; //导入依赖的package包/类
public B reduce(List<B> buckets, ReduceContext context) {
    long docCount = 0;
    // For the per term doc count error we add up the errors from the
    // shards that did not respond with the term. To do this we add up
    // the errors from the shards that did respond with the terms and
    // subtract that from the sum of the error from all shards
    long docCountError = 0;
    List<InternalAggregations> aggregationsList = new ArrayList<>(buckets.size());
    for (B bucket : buckets) {
        docCount += bucket.docCount;
        if (docCountError != -1) {
            if (bucket.docCountError == -1) {
                docCountError = -1;
            } else {
                docCountError += bucket.docCountError;
            }
        }
        aggregationsList.add(bucket.aggregations);
    }
    InternalAggregations aggs = InternalAggregations.reduce(aggregationsList, context);
    return newBucket(docCount, aggs, docCountError);
}
 

import org.elasticsearch.search.aggregations.InternalAggregations; //导入依赖的package包/类
public void testConsumerConcurrently() throws InterruptedException {
    int expectedNumResults = randomIntBetween(1, 100);
    int bufferSize = randomIntBetween(2, 200);

    SearchRequest request = new SearchRequest();
    request.source(new SearchSourceBuilder().aggregation(AggregationBuilders.avg("foo")));
    request.setBatchedReduceSize(bufferSize);
    InitialSearchPhase.SearchPhaseResults<QuerySearchResultProvider> consumer =
        searchPhaseController.newSearchPhaseResults(request, expectedNumResults);
    AtomicInteger max = new AtomicInteger();
    CountDownLatch latch = new CountDownLatch(expectedNumResults);
    for (int i = 0; i < expectedNumResults; i++) {
        int id = i;
        Thread t = new Thread(() -> {
            int number = randomIntBetween(1, 1000);
            max.updateAndGet(prev -> Math.max(prev, number));
            QuerySearchResult result = new QuerySearchResult(id, new SearchShardTarget("node", new Index("a", "b"), id));
            result.topDocs(new TopDocs(id, new ScoreDoc[0], 0.0F), new DocValueFormat[0]);
            InternalAggregations aggs = new InternalAggregations(Arrays.asList(new InternalMax("test", (double) number,
                DocValueFormat.RAW, Collections.emptyList(), Collections.emptyMap())));
            result.aggregations(aggs);
            consumer.consumeResult(id, result);
            latch.countDown();

        });
        t.start();
    }
    latch.await();
    SearchPhaseController.ReducedQueryPhase reduce = consumer.reduce();
    InternalMax internalMax = (InternalMax) reduce.aggregations.asList().get(0);
    assertEquals(max.get(), internalMax.getValue(), 0.0D);
}
 

import org.elasticsearch.search.aggregations.InternalAggregations; //导入依赖的package包/类
public B reduce(List<B> buckets, ReduceContext context) {
    long subsetDf = 0;
    long supersetDf = 0;
    List<InternalAggregations> aggregationsList = new ArrayList<>(buckets.size());
    for (B bucket : buckets) {
        subsetDf += bucket.subsetDf;
        supersetDf += bucket.supersetDf;
        aggregationsList.add(bucket.aggregations);
    }
    InternalAggregations aggs = InternalAggregations.reduce(aggregationsList, context);
    return newBucket(subsetDf, subsetSize, supersetDf, supersetSize, aggs);
}
 

import org.elasticsearch.search.aggregations.InternalAggregations; //导入依赖的package包/类
public Bucket(DocValueFormat format, boolean keyed, String key, BytesRef from, BytesRef to,
        long docCount, InternalAggregations aggregations) {
    this.format = format;
    this.keyed = keyed;
    this.key = key;
    this.from = from;
    this.to = to;
    this.docCount = docCount;
    this.aggregations = aggregations;
}
 

import org.elasticsearch.search.aggregations.InternalAggregations; //导入依赖的package包/类
public Bucket(String key, double from, double to, long docCount, InternalAggregations aggregations, boolean keyed,
        DocValueFormat formatter) {
    this(keyed, formatter);
    this.key = key != null ? key : generateKey(from, to, formatter);
    this.from = from;
    this.to = to;
    this.docCount = docCount;
    this.aggregations = aggregations;
}
 

import org.elasticsearch.search.aggregations.InternalAggregations; //导入依赖的package包/类
Bucket reduce(List<Bucket> ranges, ReduceContext context) {
    long docCount = 0;
    List<InternalAggregations> aggregationsList = new ArrayList<>(ranges.size());
    for (Bucket range : ranges) {
        docCount += range.docCount;
        aggregationsList.add(range.aggregations);
    }
    final InternalAggregations aggs = InternalAggregations.reduce(aggregationsList, context);
    return getFactory().createBucket(key, from, to, docCount, aggs, keyed, format);
}
 

import org.elasticsearch.search.aggregations.InternalAggregations; //导入依赖的package包/类
public InternalSearchResponse(InternalSearchHits hits, InternalAggregations aggregations, Suggest suggest,
                              InternalProfileShardResults profileResults, boolean timedOut, Boolean terminatedEarly) {
    this.hits = hits;
    this.aggregations = aggregations;
    this.suggest = suggest;
    this.profileResults = profileResults;
    this.timedOut = timedOut;
    this.terminatedEarly = terminatedEarly;
}
 

import org.elasticsearch.search.aggregations.InternalAggregations; //导入依赖的package包/类
public Bucket reduce(List<? extends Bucket> buckets, ReduceContext context) {
    List<InternalAggregations> aggregationsList = new ArrayList<>(buckets.size());
    long docCount = 0;
    for (Bucket bucket : buckets) {
        docCount += bucket.docCount;
        aggregationsList.add(bucket.aggregations);
    }
    final InternalAggregations aggs = InternalAggregations.reduce(aggregationsList, context);
    return new Bucket(geohashAsLong, docCount, aggs);
}
 

import org.elasticsearch.search.aggregations.InternalAggregations; //导入依赖的package包/类
public Bucket(long key, long docCount, boolean keyed, ValueFormatter formatter, Factory factory,
        InternalAggregations aggregations) {
    this(keyed, formatter, factory);
    this.key = key;
    this.docCount = docCount;
    this.aggregations = aggregations;
}
 

import org.elasticsearch.search.aggregations.InternalAggregations; //导入依赖的package包/类
/**
 * Read from a stream.
 */
protected Bucket(StreamInput in, DocValueFormat formatter, boolean showDocCountError) throws IOException {
    this.showDocCountError = showDocCountError;
    this.format = formatter;
    docCount = in.readVLong();
    docCountError = -1;
    if (showDocCountError) {
        docCountError = in.readLong();
    }
    aggregations = InternalAggregations.readAggregations(in);
}
 

import org.elasticsearch.search.aggregations.InternalAggregations; //导入依赖的package包/类
@Override
public void readFrom(StreamInput in) throws IOException {
    termBytes = in.readBytesRef();
    subsetDf = in.readVLong();
    supersetDf = in.readVLong();
    score = in.readDouble();
    aggregations = InternalAggregations.readAggregations(in);
}
 

import org.elasticsearch.search.aggregations.InternalAggregations; //导入依赖的package包/类
@Override
public void readFrom(StreamInput in) throws IOException {
    subsetDf = in.readVLong();
    supersetDf = in.readVLong();
    term = in.readLong();
    score = in.readDouble();
    aggregations = InternalAggregations.readAggregations(in);

}
 

import org.elasticsearch.search.aggregations.InternalAggregations; //导入依赖的package包/类
InternalBucket reduce(List<InternalBucket> buckets, ReduceContext context) {
    InternalBucket reduced = null;
    List<InternalAggregations> aggregationsList = new ArrayList<>(buckets.size());
    for (InternalBucket bucket : buckets) {
        if (reduced == null) {
            reduced = new InternalBucket(bucket.key, bucket.docCount, bucket.aggregations);
        } else {
            reduced.docCount += bucket.docCount;
        }
        aggregationsList.add(bucket.aggregations);
    }
    reduced.aggregations = InternalAggregations.reduce(aggregationsList, context);
    return reduced;
}
 

import org.elasticsearch.search.aggregations.InternalAggregations; //导入依赖的package包/类
@Override
public void readFrom(StreamInput in) throws IOException {
    hits = SearchHits.readSearchHits(in);
    if (in.readBoolean()) {
        aggregations = InternalAggregations.readAggregations(in);
    }
    if (in.readBoolean()) {
        suggest = Suggest.readSuggest(in);
    }
    timedOut = in.readBoolean();
    terminatedEarly = in.readOptionalBoolean();
    profileResults = in.readOptionalWriteable(SearchProfileShardResults::new);
    numReducePhases = in.readVInt();
}
 

import org.elasticsearch.search.aggregations.InternalAggregations; //导入依赖的package包/类
EmptyBucketInfo(double interval, double offset, double minBound, double maxBound, InternalAggregations subAggregations) {
    this.interval = interval;
    this.offset = offset;
    this.minBound = minBound;
    this.maxBound = maxBound;
    this.subAggregations = subAggregations;
}
 

import org.elasticsearch.search.aggregations.InternalAggregations; //导入依赖的package包/类
public Bucket(long key, long docCount, boolean keyed, DocValueFormat format,
        InternalAggregations aggregations) {
    this.format = format;
    this.keyed = keyed;
    this.key = key;
    this.docCount = docCount;
    this.aggregations = aggregations;
}
 

import org.elasticsearch.search.aggregations.InternalAggregations; //导入依赖的package包/类
Bucket reduce(List<Bucket> buckets, ReduceContext context) {
    List<InternalAggregations> aggregations = new ArrayList<>(buckets.size());
    long docCount = 0;
    for (Bucket bucket : buckets) {
        docCount += bucket.docCount;
        aggregations.add((InternalAggregations) bucket.getAggregations());
    }
    InternalAggregations aggs = InternalAggregations.reduce(aggregations, context);
    return new InternalDateHistogram.Bucket(key, docCount, keyed, format, aggs);
}