From 071d8b2c1c4f89ed62f8c8988dcd762aa70694e6 Mon Sep 17 00:00:00 2001 From: Armin Braun Date: Sat, 4 Jul 2020 14:02:16 +0200 Subject: [PATCH] Deduplicate Empty InternalAggregations (#58386) (#59032) Working through a heap dump for an unrelated issue I found that we can easily rack up tens of MBs of duplicate empty instances in some cases. I moved to a static constructor to guard against that in all cases. --- .../search/TransportNoopSearchAction.java | 2 +- .../action/search/SearchPhaseController.java | 4 +- .../search/aggregations/AggregatorBase.java | 2 +- .../aggregations/InternalAggregations.java | 38 +++++++++++-------- .../InternalMultiBucketAggregation.java | 2 +- .../bucket/BucketsAggregator.java | 13 +------ .../InternalSingleBucketAggregation.java | 4 +- .../adjacency/InternalAdjacencyMatrix.java | 2 +- .../bucket/composite/InternalComposite.java | 2 +- .../bucket/filter/InternalFilters.java | 2 +- .../bucket/geogrid/InternalGeoGridBucket.java | 2 +- .../histogram/InternalAutoDateHistogram.java | 4 +- .../histogram/InternalDateHistogram.java | 4 +- .../bucket/histogram/InternalHistogram.java | 4 +- .../InternalVariableWidthHistogram.java | 4 +- .../bucket/range/InternalBinaryRange.java | 2 +- .../bucket/range/InternalDateRange.java | 2 +- .../bucket/range/InternalGeoDistance.java | 5 --- .../bucket/range/InternalRange.java | 2 +- .../bucket/terms/InternalRareTerms.java | 2 +- .../bucket/terms/InternalTerms.java | 2 +- .../bucket/terms/SignificantLongTerms.java | 2 +- .../bucket/terms/SignificantStringTerms.java | 2 +- .../BucketScriptPipelineAggregator.java | 2 +- .../CumulativeSumPipelineAggregator.java | 2 +- .../DerivativePipelineAggregator.java | 2 +- .../pipeline/MovAvgPipelineAggregator.java | 6 +-- .../pipeline/MovFnPipelineAggregator.java | 2 +- .../SerialDiffPipelineAggregator.java | 2 +- .../pipeline/SiblingPipelineAggregator.java | 2 +- .../internal/InternalSearchResponse.java | 2 +- .../search/query/QuerySearchResult.java | 4 +- .../search/SearchPhaseControllerTests.java | 12 +++--- .../search/SearchResponseMergerTests.java | 2 +- .../aggregations/AggregationsTests.java | 2 +- .../InternalAggregationsTests.java | 6 +-- .../InternalMultiBucketAggregationTests.java | 22 +++++------ .../bucket/filter/InternalFilterTests.java | 3 +- .../bucket/filter/InternalFiltersTests.java | 3 +- .../InternalAutoDateHistogramTests.java | 8 ++-- ...ternalSingleBucketAggregationTestCase.java | 4 +- ...nternalMultiBucketAggregationTestCase.java | 2 +- ...mulativeCardinalityPipelineAggregator.java | 2 +- .../MovingPercentilesPipelineAggregator.java | 4 +- .../NormalizePipelineAggregator.java | 2 +- .../xpack/search/AsyncSearchTaskTests.java | 8 ++-- .../aggs/InferencePipelineAggregator.java | 2 +- .../rollup/RollupResponseTranslator.java | 11 ++---- .../RollupResponseTranslationTests.java | 29 +++++++------- .../rollup/action/SearchActionTests.java | 10 ++--- .../rollup/job/RollupIndexerStateTests.java | 6 +-- 51 files changed, 127 insertions(+), 143 deletions(-) diff --git a/client/client-benchmark-noop-api-plugin/src/main/java/org/elasticsearch/plugin/noop/action/search/TransportNoopSearchAction.java b/client/client-benchmark-noop-api-plugin/src/main/java/org/elasticsearch/plugin/noop/action/search/TransportNoopSearchAction.java index 745f247ad1ba5..09cce72f5f7aa 100644 --- a/client/client-benchmark-noop-api-plugin/src/main/java/org/elasticsearch/plugin/noop/action/search/TransportNoopSearchAction.java +++ b/client/client-benchmark-noop-api-plugin/src/main/java/org/elasticsearch/plugin/noop/action/search/TransportNoopSearchAction.java @@ -49,7 +49,7 @@ protected void doExecute(Task task, SearchRequest request, ActionListener aggregations) { + private InternalAggregations(List aggregations) { super(aggregations); this.pipelineTreeForBwcSerialization = null; } @@ -85,19 +85,26 @@ public InternalAggregations(List aggregations, Supplier in.readNamedWriteable(InternalAggregation.class))); + public static InternalAggregations from(List aggregations) { + if (aggregations.isEmpty()) { + return EMPTY; + } + return new InternalAggregations(aggregations); + } + + public static InternalAggregations readFrom(StreamInput in) throws IOException { + final InternalAggregations res = from(in.readList(stream -> in.readNamedWriteable(InternalAggregation.class))); if (in.getVersion().before(Version.V_7_8_0) && in.getVersion().onOrAfter(Version.V_6_7_0)) { - in.readNamedWriteableList(PipelineAggregator.class); + /* + * Setting the pipeline tree source to null is here is correct but + * only because we don't immediately pass the InternalAggregations + * off to another node. Instead, we always reduce together with + * many aggregations and that always adds the tree read from the + * current request. + */ + in.readNamedWriteableList(PipelineAggregator.class); } - /* - * Setting the pipeline tree source to null is here is correct but - * only because we don't immediately pass the InternalAggregations - * off to another node. Instead, we always reduce together with - * many aggregations and that always adds the tree read from the - * current request. - */ - pipelineTreeForBwcSerialization = null; + return res; } @Override @@ -206,10 +213,10 @@ public static InternalAggregations topLevelReduce(List agg for (PipelineAggregator pipelineAggregator : context.pipelineTreeRoot().aggregators()) { SiblingPipelineAggregator sib = (SiblingPipelineAggregator) pipelineAggregator; - InternalAggregation newAgg = sib.doReduce(new InternalAggregations(reducedInternalAggs), context); + InternalAggregation newAgg = sib.doReduce(from(reducedInternalAggs), context); reducedInternalAggs.add(newAgg); } - return new InternalAggregations(reducedInternalAggs); + return from(reducedInternalAggs); } return reduced; } @@ -258,7 +265,6 @@ public static InternalAggregations reduce(List aggregation * Version of {@link #reduce(List, ReduceContext, Function)} for nodes inside the aggregation tree. */ public static InternalAggregations reduce(List aggregationsList, ReduceContext context) { - return reduce(aggregationsList, context, InternalAggregations::new); + return reduce(aggregationsList, context, InternalAggregations::from); } - } diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/InternalMultiBucketAggregation.java b/server/src/main/java/org/elasticsearch/search/aggregations/InternalMultiBucketAggregation.java index 8b69573f14fa4..a7333f124a4ea 100644 --- a/server/src/main/java/org/elasticsearch/search/aggregations/InternalMultiBucketAggregation.java +++ b/server/src/main/java/org/elasticsearch/search/aggregations/InternalMultiBucketAggregation.java @@ -192,7 +192,7 @@ private List reducePipelineBuckets(ReduceContext reduceContext, PipelineTree PipelineTree subTree = pipelineTree.subTree(agg.getName()); aggs.add(((InternalAggregation)agg).reducePipelines((InternalAggregation)agg, reduceContext, subTree)); } - reducedBuckets.add(createBucket(new InternalAggregations(aggs), bucket)); + reducedBuckets.add(createBucket(InternalAggregations.from(aggs), bucket)); } return reducedBuckets; } diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/BucketsAggregator.java b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/BucketsAggregator.java index 963e9448c15f0..3c50fb0f64171 100644 --- a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/BucketsAggregator.java +++ b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/BucketsAggregator.java @@ -178,7 +178,7 @@ protected final InternalAggregations[] buildSubAggsForBuckets(long[] bucketOrdsT slice[i] = aggregations[i][ord]; } final int thisOrd = ord; - result[ord] = new InternalAggregations(new AbstractList() { + result[ord] = InternalAggregations.from(new AbstractList() { @Override public InternalAggregation get(int index) { return aggregations[index][thisOrd]; @@ -353,17 +353,6 @@ protected interface ResultBuilderForVariable { InternalAggregation build(long owninigBucketOrd, List buckets); } - /** - * Utility method to build empty aggregations of the sub aggregators. - */ - protected final InternalAggregations bucketEmptyAggregations() { - final InternalAggregation[] aggregations = new InternalAggregation[subAggregators.length]; - for (int i = 0; i < subAggregators.length; i++) { - aggregations[i] = subAggregators[i].buildEmptyAggregation(); - } - return new InternalAggregations(Arrays.asList(aggregations)); - } - @Override public final void close() { try (Releasable releasable = docCounts) { diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/InternalSingleBucketAggregation.java b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/InternalSingleBucketAggregation.java index 5d79ab2ce805e..ff5eaa33661f4 100644 --- a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/InternalSingleBucketAggregation.java +++ b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/InternalSingleBucketAggregation.java @@ -64,7 +64,7 @@ protected InternalSingleBucketAggregation(String name, long docCount, InternalAg protected InternalSingleBucketAggregation(StreamInput in) throws IOException { super(in); docCount = in.readVLong(); - aggregations = new InternalAggregations(in); + aggregations = InternalAggregations.readFrom(in); } @Override @@ -128,7 +128,7 @@ public final InternalAggregation reducePipelines( PipelineTree subTree = pipelineTree.subTree(agg.getName()); aggs.add(((InternalAggregation)agg).reducePipelines((InternalAggregation)agg, reduceContext, subTree)); } - InternalAggregations reducedSubAggs = new InternalAggregations(aggs); + InternalAggregations reducedSubAggs = InternalAggregations.from(aggs); reduced = create(reducedSubAggs); } return super.reducePipelines(reduced, reduceContext, pipelineTree); diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/adjacency/InternalAdjacencyMatrix.java b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/adjacency/InternalAdjacencyMatrix.java index 6a9d51c3ef32f..60d97aabac4ff 100644 --- a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/adjacency/InternalAdjacencyMatrix.java +++ b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/adjacency/InternalAdjacencyMatrix.java @@ -56,7 +56,7 @@ public InternalBucket(String key, long docCount, InternalAggregations aggregatio public InternalBucket(StreamInput in) throws IOException { key = in.readOptionalString(); docCount = in.readVLong(); - aggregations = new InternalAggregations(in); + aggregations = InternalAggregations.readFrom(in); } @Override diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/composite/InternalComposite.java b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/composite/InternalComposite.java index db2dc5e34d79e..d37a5d72a8ade 100644 --- a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/composite/InternalComposite.java +++ b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/composite/InternalComposite.java @@ -301,7 +301,7 @@ public static class InternalBucket extends InternalMultiBucketAggregation.Intern InternalBucket(StreamInput in, List sourceNames, List formats, int[] reverseMuls) throws IOException { this.key = new CompositeKey(in); this.docCount = in.readVLong(); - this.aggregations = new InternalAggregations(in); + this.aggregations = InternalAggregations.readFrom(in); this.reverseMuls = reverseMuls; this.sourceNames = sourceNames; this.formats = formats; diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/filter/InternalFilters.java b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/filter/InternalFilters.java index c7e359c342051..de16dac7ea902 100644 --- a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/filter/InternalFilters.java +++ b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/filter/InternalFilters.java @@ -56,7 +56,7 @@ public InternalBucket(StreamInput in, boolean keyed) throws IOException { this.keyed = keyed; key = in.readOptionalString(); docCount = in.readVLong(); - aggregations = new InternalAggregations(in); + aggregations = InternalAggregations.readFrom(in); } @Override diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/geogrid/InternalGeoGridBucket.java b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/geogrid/InternalGeoGridBucket.java index 0df9661aa9ec4..8f7dbfd2f2e2c 100644 --- a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/geogrid/InternalGeoGridBucket.java +++ b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/geogrid/InternalGeoGridBucket.java @@ -50,7 +50,7 @@ public InternalGeoGridBucket(long hashAsLong, long docCount, InternalAggregation public InternalGeoGridBucket(StreamInput in) throws IOException { hashAsLong = in.readLong(); docCount = in.readVLong(); - aggregations = new InternalAggregations(in); + aggregations = InternalAggregations.readFrom(in); } @Override diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/InternalAutoDateHistogram.java b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/InternalAutoDateHistogram.java index a7e333770672f..957112581a51f 100644 --- a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/InternalAutoDateHistogram.java +++ b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/InternalAutoDateHistogram.java @@ -72,7 +72,7 @@ public Bucket(StreamInput in, DocValueFormat format) throws IOException { this.format = format; key = in.readLong(); docCount = in.readVLong(); - aggregations = new InternalAggregations(in); + aggregations = InternalAggregations.readFrom(in); } @Override @@ -163,7 +163,7 @@ static class BucketInfo { roundingInfos[i] = new RoundingInfo(in); } roundingIdx = in.readVInt(); - emptySubAggregations = new InternalAggregations(in); + emptySubAggregations = InternalAggregations.readFrom(in); } void writeTo(StreamOutput out) throws IOException { diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/InternalDateHistogram.java b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/InternalDateHistogram.java index d40b71bb2bc35..60d1e4eeded63 100644 --- a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/InternalDateHistogram.java +++ b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/InternalDateHistogram.java @@ -76,7 +76,7 @@ public Bucket(StreamInput in, boolean keyed, DocValueFormat format) throws IOExc this.keyed = keyed; key = in.readLong(); docCount = in.readVLong(); - aggregations = new InternalAggregations(in); + aggregations = InternalAggregations.readFrom(in); } @Override @@ -174,7 +174,7 @@ static class EmptyBucketInfo { EmptyBucketInfo(StreamInput in) throws IOException { rounding = Rounding.read(in); - subAggregations = new InternalAggregations(in); + subAggregations = InternalAggregations.readFrom(in); bounds = in.readOptionalWriteable(ExtendedBounds::new); } diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/InternalHistogram.java b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/InternalHistogram.java index ac31c12c99435..73aab00c25384 100644 --- a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/InternalHistogram.java +++ b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/InternalHistogram.java @@ -72,7 +72,7 @@ public Bucket(StreamInput in, boolean keyed, DocValueFormat format) throws IOExc this.keyed = keyed; key = in.readDouble(); docCount = in.readVLong(); - aggregations = new InternalAggregations(in); + aggregations = InternalAggregations.readFrom(in); } @Override @@ -166,7 +166,7 @@ static class EmptyBucketInfo { } EmptyBucketInfo(StreamInput in) throws IOException { - this(in.readDouble(), in.readDouble(), in.readDouble(), in.readDouble(), new InternalAggregations(in)); + this(in.readDouble(), in.readDouble(), in.readDouble(), in.readDouble(), InternalAggregations.readFrom(in)); } public void writeTo(StreamOutput out) throws IOException { diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/InternalVariableWidthHistogram.java b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/InternalVariableWidthHistogram.java index ced1a4915bfdf..668cf9b5990cc 100644 --- a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/InternalVariableWidthHistogram.java +++ b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/InternalVariableWidthHistogram.java @@ -105,7 +105,7 @@ public Bucket(StreamInput in, DocValueFormat format) throws IOException { centroid = in.readDouble(); docCount = in.readVLong(); bounds = new BucketBounds(in); - aggregations = new InternalAggregations(in); + aggregations = InternalAggregations.readFrom(in); } @Override @@ -207,7 +207,7 @@ static class EmptyBucketInfo { } EmptyBucketInfo(StreamInput in) throws IOException { - this(new InternalAggregations(in)); + this(InternalAggregations.readFrom(in)); } public void writeTo(StreamOutput out) throws IOException { diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/range/InternalBinaryRange.java b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/range/InternalBinaryRange.java index 2e3f078f39817..0a08e106432e3 100644 --- a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/range/InternalBinaryRange.java +++ b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/range/InternalBinaryRange.java @@ -81,7 +81,7 @@ private static Bucket createFromStream(StreamInput in, DocValueFormat format, bo BytesRef from = in.readBoolean() ? in.readBytesRef() : null; BytesRef to = in.readBoolean() ? in.readBytesRef() : null; long docCount = in.readLong(); - InternalAggregations aggregations = new InternalAggregations(in); + InternalAggregations aggregations = InternalAggregations.readFrom(in); return new Bucket(format, keyed, key, from, to, docCount, aggregations); } diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/range/InternalDateRange.java b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/range/InternalDateRange.java index a34a3f2e57d93..2c937ab104c54 100644 --- a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/range/InternalDateRange.java +++ b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/range/InternalDateRange.java @@ -37,7 +37,7 @@ public static class Bucket extends InternalRange.Bucket { public Bucket(String key, double from, double to, long docCount, List aggregations, boolean keyed, DocValueFormat formatter) { - super(key, from, to, docCount, new InternalAggregations(aggregations), keyed, formatter); + super(key, from, to, docCount, InternalAggregations.from(aggregations), keyed, formatter); } public Bucket(String key, double from, double to, long docCount, InternalAggregations aggregations, boolean keyed, diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/range/InternalGeoDistance.java b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/range/InternalGeoDistance.java index 14f0b097719fe..6699d1d1c48ef 100644 --- a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/range/InternalGeoDistance.java +++ b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/range/InternalGeoDistance.java @@ -20,7 +20,6 @@ import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.search.DocValueFormat; -import org.elasticsearch.search.aggregations.InternalAggregation; import org.elasticsearch.search.aggregations.InternalAggregations; import org.elasticsearch.search.aggregations.support.CoreValuesSourceType; import org.elasticsearch.search.aggregations.support.ValueType; @@ -35,10 +34,6 @@ public class InternalGeoDistance extends InternalRange aggregations, boolean keyed) { - this(key, from, to, docCount, new InternalAggregations(aggregations), keyed); - } - Bucket(String key, double from, double to, long docCount, InternalAggregations aggregations, boolean keyed) { super(key, from, to, docCount, aggregations, keyed, DocValueFormat.RAW); } diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/range/InternalRange.java b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/range/InternalRange.java index af185bed29cd0..a8c61c0ad368b 100644 --- a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/range/InternalRange.java +++ b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/range/InternalRange.java @@ -248,7 +248,7 @@ public InternalRange(StreamInput in) throws IOException { ? in.readString() : in.readOptionalString(); ranges.add(getFactory().createBucket(key, in.readDouble(), in.readDouble(), in.readVLong(), - new InternalAggregations(in), keyed, format)); + InternalAggregations.readFrom(in), keyed, format)); } this.ranges = ranges; } diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/terms/InternalRareTerms.java b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/terms/InternalRareTerms.java index 3ae5f7f2b5d9b..2d04eda164021 100644 --- a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/terms/InternalRareTerms.java +++ b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/terms/InternalRareTerms.java @@ -68,7 +68,7 @@ protected Bucket(long docCount, InternalAggregations aggregations, DocValueForma protected Bucket(StreamInput in, DocValueFormat formatter) throws IOException { this.format = formatter; docCount = in.readVLong(); - aggregations = new InternalAggregations(in); + aggregations = InternalAggregations.readFrom(in); } @Override diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/terms/InternalTerms.java b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/terms/InternalTerms.java index 05303ffad2fb0..a0ef8861b7ba7 100644 --- a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/terms/InternalTerms.java +++ b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/terms/InternalTerms.java @@ -84,7 +84,7 @@ protected Bucket(StreamInput in, DocValueFormat formatter, boolean showDocCountE if (showDocCountError) { docCountError = in.readLong(); } - aggregations = new InternalAggregations(in); + aggregations = InternalAggregations.readFrom(in); } @Override diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/terms/SignificantLongTerms.java b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/terms/SignificantLongTerms.java index 2cedc09bbee10..d047d26127868 100644 --- a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/terms/SignificantLongTerms.java +++ b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/terms/SignificantLongTerms.java @@ -53,7 +53,7 @@ static class Bucket extends InternalSignificantTerms.Bucket { supersetDf = in.readVLong(); term = in.readLong(); score = in.readDouble(); - aggregations = new InternalAggregations(in); + aggregations = InternalAggregations.readFrom(in); } @Override diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/terms/SignificantStringTerms.java b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/terms/SignificantStringTerms.java index 8053b540e5830..ab6de41f3cb64 100644 --- a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/terms/SignificantStringTerms.java +++ b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/terms/SignificantStringTerms.java @@ -57,7 +57,7 @@ public Bucket(StreamInput in, long subsetSize, long supersetSize, DocValueFormat subsetDf = in.readVLong(); supersetDf = in.readVLong(); score = in.readDouble(); - aggregations = new InternalAggregations(in); + aggregations = InternalAggregations.readFrom(in); } @Override diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/pipeline/BucketScriptPipelineAggregator.java b/server/src/main/java/org/elasticsearch/search/aggregations/pipeline/BucketScriptPipelineAggregator.java index 7d9a0395616ed..de1a430243594 100644 --- a/server/src/main/java/org/elasticsearch/search/aggregations/pipeline/BucketScriptPipelineAggregator.java +++ b/server/src/main/java/org/elasticsearch/search/aggregations/pipeline/BucketScriptPipelineAggregator.java @@ -117,7 +117,7 @@ public InternalAggregation reduce(InternalAggregation aggregation, ReduceContext InternalSimpleValue simpleValue = new InternalSimpleValue(name(), returned.doubleValue(), formatter, metadata()); aggs.add(simpleValue); - InternalMultiBucketAggregation.InternalBucket newBucket = originalAgg.createBucket(new InternalAggregations(aggs), + InternalMultiBucketAggregation.InternalBucket newBucket = originalAgg.createBucket(InternalAggregations.from(aggs), bucket); newBuckets.add(newBucket); } diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/pipeline/CumulativeSumPipelineAggregator.java b/server/src/main/java/org/elasticsearch/search/aggregations/pipeline/CumulativeSumPipelineAggregator.java index 5af6fb5e36ca9..8f78cbbb184d0 100644 --- a/server/src/main/java/org/elasticsearch/search/aggregations/pipeline/CumulativeSumPipelineAggregator.java +++ b/server/src/main/java/org/elasticsearch/search/aggregations/pipeline/CumulativeSumPipelineAggregator.java @@ -87,7 +87,7 @@ public InternalAggregation reduce(InternalAggregation aggregation, ReduceContext .map((p) -> (InternalAggregation) p) .collect(Collectors.toList()); aggs.add(new InternalSimpleValue(name(), sum, formatter, metadata())); - Bucket newBucket = factory.createBucket(factory.getKey(bucket), bucket.getDocCount(), new InternalAggregations(aggs)); + Bucket newBucket = factory.createBucket(factory.getKey(bucket), bucket.getDocCount(), InternalAggregations.from(aggs)); newBuckets.add(newBucket); } return factory.createAggregation(newBuckets); diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/pipeline/DerivativePipelineAggregator.java b/server/src/main/java/org/elasticsearch/search/aggregations/pipeline/DerivativePipelineAggregator.java index d603202586d6f..68125b44f0e99 100644 --- a/server/src/main/java/org/elasticsearch/search/aggregations/pipeline/DerivativePipelineAggregator.java +++ b/server/src/main/java/org/elasticsearch/search/aggregations/pipeline/DerivativePipelineAggregator.java @@ -98,7 +98,7 @@ public InternalAggregation reduce(InternalAggregation aggregation, ReduceContext return (InternalAggregation) p; }).collect(Collectors.toList()); aggs.add(new InternalDerivative(name(), gradient, xDiff, formatter, metadata())); - Bucket newBucket = factory.createBucket(factory.getKey(bucket), bucket.getDocCount(), new InternalAggregations(aggs)); + Bucket newBucket = factory.createBucket(factory.getKey(bucket), bucket.getDocCount(), InternalAggregations.from(aggs)); newBuckets.add(newBucket); } else { newBuckets.add(bucket); diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/pipeline/MovAvgPipelineAggregator.java b/server/src/main/java/org/elasticsearch/search/aggregations/pipeline/MovAvgPipelineAggregator.java index 38794ae704215..a696efafeecc8 100644 --- a/server/src/main/java/org/elasticsearch/search/aggregations/pipeline/MovAvgPipelineAggregator.java +++ b/server/src/main/java/org/elasticsearch/search/aggregations/pipeline/MovAvgPipelineAggregator.java @@ -127,7 +127,7 @@ public InternalAggregation reduce(InternalAggregation aggregation, ReduceContext .map((p) -> (InternalAggregation) p) .collect(Collectors.toList()); aggs.add(new InternalSimpleValue(name(), movavg, formatter, metadata())); - newBucket = factory.createBucket(factory.getKey(bucket), bucket.getDocCount(), new InternalAggregations(aggs)); + newBucket = factory.createBucket(factory.getKey(bucket), bucket.getDocCount(), InternalAggregations.from(aggs)); } if (predict > 0) { @@ -158,7 +158,7 @@ public InternalAggregation reduce(InternalAggregation aggregation, ReduceContext .collect(Collectors.toList()); aggs.add(new InternalSimpleValue(name(), predictions[i], formatter, metadata())); - Bucket newBucket = factory.createBucket(newKey, bucket.getDocCount(), new InternalAggregations(aggs)); + Bucket newBucket = factory.createBucket(newKey, bucket.getDocCount(), InternalAggregations.from(aggs)); // Overwrite the existing bucket with the new version newBuckets.set(lastValidPosition + i + 1, newBucket); @@ -168,7 +168,7 @@ public InternalAggregation reduce(InternalAggregation aggregation, ReduceContext aggs = new ArrayList<>(); aggs.add(new InternalSimpleValue(name(), predictions[i], formatter, metadata())); - Bucket newBucket = factory.createBucket(newKey, 0, new InternalAggregations(aggs)); + Bucket newBucket = factory.createBucket(newKey, 0, InternalAggregations.from(aggs)); // Since this is a new bucket, simply append it newBuckets.add(newBucket); diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/pipeline/MovFnPipelineAggregator.java b/server/src/main/java/org/elasticsearch/search/aggregations/pipeline/MovFnPipelineAggregator.java index 00fef79daebd2..1db9bb5f1a162 100644 --- a/server/src/main/java/org/elasticsearch/search/aggregations/pipeline/MovFnPipelineAggregator.java +++ b/server/src/main/java/org/elasticsearch/search/aggregations/pipeline/MovFnPipelineAggregator.java @@ -157,7 +157,7 @@ public InternalAggregation reduce(InternalAggregation aggregation, InternalAggre .map(InternalAggregation.class::cast) .collect(Collectors.toList()); aggs.add(new InternalSimpleValue(name(), movavg, formatter, metadata())); - newBucket = factory.createBucket(factory.getKey(bucket), bucket.getDocCount(), new InternalAggregations(aggs)); + newBucket = factory.createBucket(factory.getKey(bucket), bucket.getDocCount(), InternalAggregations.from(aggs)); index++; } newBuckets.add(newBucket); diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/pipeline/SerialDiffPipelineAggregator.java b/server/src/main/java/org/elasticsearch/search/aggregations/pipeline/SerialDiffPipelineAggregator.java index 6e0c2d8aa995b..6a30fa494f5a9 100644 --- a/server/src/main/java/org/elasticsearch/search/aggregations/pipeline/SerialDiffPipelineAggregator.java +++ b/server/src/main/java/org/elasticsearch/search/aggregations/pipeline/SerialDiffPipelineAggregator.java @@ -114,7 +114,7 @@ public InternalAggregation reduce(InternalAggregation aggregation, ReduceContext List aggs = StreamSupport.stream(bucket.getAggregations().spliterator(), false).map( (p) -> (InternalAggregation) p).collect(Collectors.toList()); aggs.add(new InternalSimpleValue(name(), diff, formatter, metadata())); - newBucket = factory.createBucket(factory.getKey(bucket), bucket.getDocCount(), new InternalAggregations(aggs)); + newBucket = factory.createBucket(factory.getKey(bucket), bucket.getDocCount(), InternalAggregations.from(aggs)); } newBuckets.add(newBucket); diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/pipeline/SiblingPipelineAggregator.java b/server/src/main/java/org/elasticsearch/search/aggregations/pipeline/SiblingPipelineAggregator.java index 794dc4d85693d..073294a44c7c3 100644 --- a/server/src/main/java/org/elasticsearch/search/aggregations/pipeline/SiblingPipelineAggregator.java +++ b/server/src/main/java/org/elasticsearch/search/aggregations/pipeline/SiblingPipelineAggregator.java @@ -46,7 +46,7 @@ public InternalAggregation reduce(InternalAggregation aggregation, ReduceContext return aggregation.copyWithRewritenBuckets(aggregations -> { List aggs = aggregations.copyResults(); aggs.add(doReduce(aggregations, reduceContext)); - return new InternalAggregations(aggs); + return InternalAggregations.from(aggs); }); } diff --git a/server/src/main/java/org/elasticsearch/search/internal/InternalSearchResponse.java b/server/src/main/java/org/elasticsearch/search/internal/InternalSearchResponse.java index 323c41c110f93..5cbdf92e00b72 100644 --- a/server/src/main/java/org/elasticsearch/search/internal/InternalSearchResponse.java +++ b/server/src/main/java/org/elasticsearch/search/internal/InternalSearchResponse.java @@ -52,7 +52,7 @@ public InternalSearchResponse(SearchHits hits, InternalAggregations aggregations public InternalSearchResponse(StreamInput in) throws IOException { super( new SearchHits(in), - in.readBoolean() ? new InternalAggregations(in) : null, + in.readBoolean() ? InternalAggregations.readFrom(in) : null, in.readBoolean() ? new Suggest(in) : null, in.readBoolean(), in.readOptionalBoolean(), diff --git a/server/src/main/java/org/elasticsearch/search/query/QuerySearchResult.java b/server/src/main/java/org/elasticsearch/search/query/QuerySearchResult.java index e1d1d702aad30..6309893c95e8c 100644 --- a/server/src/main/java/org/elasticsearch/search/query/QuerySearchResult.java +++ b/server/src/main/java/org/elasticsearch/search/query/QuerySearchResult.java @@ -319,7 +319,7 @@ public void readFromWithId(SearchContextId id, StreamInput in) throws IOExceptio setTopDocs(readTopDocs(in)); if (in.getVersion().before(Version.V_7_7_0)) { if (hasAggs = in.readBoolean()) { - aggregations = DelayableWriteable.referencing(new InternalAggregations(in)); + aggregations = DelayableWriteable.referencing(InternalAggregations.readFrom(in)); } if (in.getVersion().before(Version.V_7_2_0)) { // The list of PipelineAggregators is sent by old versions. We don't need it anyway. @@ -327,7 +327,7 @@ public void readFromWithId(SearchContextId id, StreamInput in) throws IOExceptio } } else { if (hasAggs = in.readBoolean()) { - aggregations = DelayableWriteable.delayed(InternalAggregations::new, in); + aggregations = DelayableWriteable.delayed(InternalAggregations::readFrom, in); } } if (in.readBoolean()) { diff --git a/server/src/test/java/org/elasticsearch/action/search/SearchPhaseControllerTests.java b/server/src/test/java/org/elasticsearch/action/search/SearchPhaseControllerTests.java index e8abe21bfc280..7b93b9aa25e90 100644 --- a/server/src/test/java/org/elasticsearch/action/search/SearchPhaseControllerTests.java +++ b/server/src/test/java/org/elasticsearch/action/search/SearchPhaseControllerTests.java @@ -396,7 +396,7 @@ private void consumerTestCase(int numEmptyResponses) { new SearchShardTarget("node", new ShardId("a", "b", 0), null, OriginalIndices.NONE)); result.topDocs(new TopDocsAndMaxScore(new TopDocs(new TotalHits(0, TotalHits.Relation.EQUAL_TO), new ScoreDoc[0]), Float.NaN), new DocValueFormat[0]); - InternalAggregations aggs = new InternalAggregations(singletonList(new InternalMax("test", 1.0D, DocValueFormat.RAW, emptyMap()))); + InternalAggregations aggs = InternalAggregations.from(singletonList(new InternalMax("test", 1.0D, DocValueFormat.RAW, emptyMap()))); result.aggregations(aggs); result.setShardIndex(0); consumer.consumeResult(result); @@ -405,7 +405,7 @@ private void consumerTestCase(int numEmptyResponses) { new SearchShardTarget("node", new ShardId("a", "b", 0), null, OriginalIndices.NONE)); result.topDocs(new TopDocsAndMaxScore(new TopDocs(new TotalHits(0, TotalHits.Relation.EQUAL_TO), new ScoreDoc[0]), Float.NaN), new DocValueFormat[0]); - aggs = new InternalAggregations(singletonList(new InternalMax("test", 3.0D, DocValueFormat.RAW, emptyMap()))); + aggs = InternalAggregations.from(singletonList(new InternalMax("test", 3.0D, DocValueFormat.RAW, emptyMap()))); result.aggregations(aggs); result.setShardIndex(2); consumer.consumeResult(result); @@ -414,7 +414,7 @@ private void consumerTestCase(int numEmptyResponses) { new SearchShardTarget("node", new ShardId("a", "b", 0), null, OriginalIndices.NONE)); result.topDocs(new TopDocsAndMaxScore(new TopDocs(new TotalHits(0, TotalHits.Relation.EQUAL_TO), new ScoreDoc[0]), Float.NaN), new DocValueFormat[0]); - aggs = new InternalAggregations(singletonList(new InternalMax("test", 2.0D, DocValueFormat.RAW, emptyMap()))); + aggs = InternalAggregations.from(singletonList(new InternalMax("test", 2.0D, DocValueFormat.RAW, emptyMap()))); result.aggregations(aggs); result.setShardIndex(1); consumer.consumeResult(result); @@ -483,7 +483,7 @@ public void testConsumerConcurrently() throws InterruptedException { result.topDocs(new TopDocsAndMaxScore( new TopDocs(new TotalHits(1, TotalHits.Relation.EQUAL_TO), new ScoreDoc[] {new ScoreDoc(0, number)}), number), new DocValueFormat[0]); - InternalAggregations aggs = new InternalAggregations(Collections.singletonList(new InternalMax("test", (double) number, + InternalAggregations aggs = InternalAggregations.from(Collections.singletonList(new InternalMax("test", (double) number, DocValueFormat.RAW, Collections.emptyMap()))); result.aggregations(aggs); result.setShardIndex(id); @@ -526,7 +526,7 @@ public void testConsumerOnlyAggs() { new SearchShardTarget("node", new ShardId("a", "b", i), null, OriginalIndices.NONE)); result.topDocs(new TopDocsAndMaxScore(new TopDocs(new TotalHits(1, TotalHits.Relation.EQUAL_TO), new ScoreDoc[0]), number), new DocValueFormat[0]); - InternalAggregations aggs = new InternalAggregations(Collections.singletonList(new InternalMax("test", (double) number, + InternalAggregations aggs = InternalAggregations.from(Collections.singletonList(new InternalMax("test", (double) number, DocValueFormat.RAW, Collections.emptyMap()))); result.aggregations(aggs); result.setShardIndex(i); @@ -878,7 +878,7 @@ public void onFinalReduce(List shards, TotalHits totalHits, Interna result.topDocs(new TopDocsAndMaxScore( new TopDocs(new TotalHits(1, TotalHits.Relation.EQUAL_TO), new ScoreDoc[]{new ScoreDoc(0, number)}), number), new DocValueFormat[0]); - InternalAggregations aggs = new InternalAggregations(Collections.singletonList( + InternalAggregations aggs = InternalAggregations.from(Collections.singletonList( new InternalMax("test", (double) number, DocValueFormat.RAW, Collections.emptyMap()))); result.aggregations(aggs); result.setShardIndex(id); diff --git a/server/src/test/java/org/elasticsearch/action/search/SearchResponseMergerTests.java b/server/src/test/java/org/elasticsearch/action/search/SearchResponseMergerTests.java index 978a4c2a77a15..1d32b10a0cc56 100644 --- a/server/src/test/java/org/elasticsearch/action/search/SearchResponseMergerTests.java +++ b/server/src/test/java/org/elasticsearch/action/search/SearchResponseMergerTests.java @@ -370,7 +370,7 @@ public void testMergeAggs() throws InterruptedException { InternalDateRange.Bucket bucket = factory.createBucket("bucket", 0, 10000, count, InternalAggregations.EMPTY, false, DocValueFormat.RAW); InternalDateRange range = factory.create(rangeAggName, singletonList(bucket), DocValueFormat.RAW, false, emptyMap()); - InternalAggregations aggs = new InternalAggregations(Arrays.asList(range, max)); + InternalAggregations aggs = InternalAggregations.from(Arrays.asList(range, max)); SearchHits searchHits = new SearchHits(new SearchHit[0], null, Float.NaN); InternalSearchResponse internalSearchResponse = new InternalSearchResponse(searchHits, aggs, null, null, false, null, 1); SearchResponse searchResponse = new SearchResponse(internalSearchResponse, null, 1, 1, 0, randomLong(), diff --git a/server/src/test/java/org/elasticsearch/search/aggregations/AggregationsTests.java b/server/src/test/java/org/elasticsearch/search/aggregations/AggregationsTests.java index d574e254c5469..26166e3b23522 100644 --- a/server/src/test/java/org/elasticsearch/search/aggregations/AggregationsTests.java +++ b/server/src/test/java/org/elasticsearch/search/aggregations/AggregationsTests.java @@ -303,6 +303,6 @@ private static InternalAggregations createTestInstance(final int minNumAggs, fin } aggs.add(testCase.createTestInstance()); } - return new InternalAggregations(aggs); + return InternalAggregations.from(aggs); } } diff --git a/server/src/test/java/org/elasticsearch/search/aggregations/InternalAggregationsTests.java b/server/src/test/java/org/elasticsearch/search/aggregations/InternalAggregationsTests.java index 0b3322cf48c88..ced6967233e55 100644 --- a/server/src/test/java/org/elasticsearch/search/aggregations/InternalAggregationsTests.java +++ b/server/src/test/java/org/elasticsearch/search/aggregations/InternalAggregationsTests.java @@ -63,7 +63,7 @@ public void testReduceEmptyAggs() { public void testNonFinalReduceTopLevelPipelineAggs() { InternalAggregation terms = new StringTerms("name", BucketOrder.key(true), 10, 1, Collections.emptyMap(), DocValueFormat.RAW, 25, false, 10, Collections.emptyList(), 0); - List aggs = singletonList(new InternalAggregations(Collections.singletonList(terms))); + List aggs = singletonList(InternalAggregations.from(Collections.singletonList(terms))); InternalAggregations reducedAggs = InternalAggregations.topLevelReduce(aggs, maxBucketReduceContext().forPartialReduction()); assertEquals(1, reducedAggs.getTopLevelPipelineAggregators().size()); assertEquals(1, reducedAggs.aggregations.size()); @@ -73,7 +73,7 @@ public void testFinalReduceTopLevelPipelineAggs() { InternalAggregation terms = new StringTerms("name", BucketOrder.key(true), 10, 1, Collections.emptyMap(), DocValueFormat.RAW, 25, false, 10, Collections.emptyList(), 0); - InternalAggregations aggs = new InternalAggregations(Collections.singletonList(terms)); + InternalAggregations aggs = InternalAggregations.from(Collections.singletonList(terms)); InternalAggregations reducedAggs = InternalAggregations.topLevelReduce(Collections.singletonList(aggs), maxBucketReduceContext().forFinalReduction()); assertEquals(0, reducedAggs.getTopLevelPipelineAggregators().size()); @@ -145,7 +145,7 @@ private void writeToAndReadFrom(InternalAggregations aggregations, int iteration aggregations.writeTo(out); try (StreamInput in = new NamedWriteableAwareStreamInput(StreamInput.wrap(out.bytes().toBytesRef().bytes), registry)) { in.setVersion(version); - InternalAggregations deserialized = new InternalAggregations(in); + InternalAggregations deserialized = InternalAggregations.readFrom(in); assertEquals(aggregations.aggregations, deserialized.aggregations); if (iteration < 2) { /* diff --git a/server/src/test/java/org/elasticsearch/search/aggregations/InternalMultiBucketAggregationTests.java b/server/src/test/java/org/elasticsearch/search/aggregations/InternalMultiBucketAggregationTests.java index 47502f1eef074..0ad6b2b2c191c 100644 --- a/server/src/test/java/org/elasticsearch/search/aggregations/InternalMultiBucketAggregationTests.java +++ b/server/src/test/java/org/elasticsearch/search/aggregations/InternalMultiBucketAggregationTests.java @@ -42,7 +42,7 @@ public void testResolveToAgg() { AggregationPath path = AggregationPath.parse("the_avg"); List buckets = new ArrayList<>(); InternalAggregation agg = new InternalAvg("the_avg", 2, 1, DocValueFormat.RAW, Collections.emptyMap()); - InternalAggregations internalAggregations = new InternalAggregations(Collections.singletonList(agg)); + InternalAggregations internalAggregations = InternalAggregations.from(Collections.singletonList(agg)); LongTerms.Bucket bucket = new LongTerms.Bucket(1, 1, internalAggregations, false, 0, DocValueFormat.RAW); buckets.add(bucket); @@ -55,7 +55,7 @@ public void testResolveToAggValue() { AggregationPath path = AggregationPath.parse("the_avg.value"); List buckets = new ArrayList<>(); InternalAggregation agg = new InternalAvg("the_avg", 2, 1, DocValueFormat.RAW, Collections.emptyMap()); - InternalAggregations internalAggregations = new InternalAggregations(Collections.singletonList(agg)); + InternalAggregations internalAggregations = InternalAggregations.from(Collections.singletonList(agg)); LongTerms.Bucket bucket = new LongTerms.Bucket(1, 1, internalAggregations, false, 0, DocValueFormat.RAW); buckets.add(bucket); @@ -68,7 +68,7 @@ public void testResolveToNothing() { AggregationPath path = AggregationPath.parse("foo.value"); List buckets = new ArrayList<>(); InternalAggregation agg = new InternalAvg("the_avg", 2, 1, DocValueFormat.RAW, Collections.emptyMap()); - InternalAggregations internalAggregations = new InternalAggregations(Collections.singletonList(agg)); + InternalAggregations internalAggregations = InternalAggregations.from(Collections.singletonList(agg)); LongTerms.Bucket bucket = new LongTerms.Bucket(1, 1, internalAggregations, false, 0, DocValueFormat.RAW); buckets.add(bucket); @@ -82,7 +82,7 @@ public void testResolveToUnknown() { AggregationPath path = AggregationPath.parse("the_avg.unknown"); List buckets = new ArrayList<>(); InternalAggregation agg = new InternalAvg("the_avg", 2, 1, DocValueFormat.RAW, Collections.emptyMap()); - InternalAggregations internalAggregations = new InternalAggregations(Collections.singletonList(agg)); + InternalAggregations internalAggregations = InternalAggregations.from(Collections.singletonList(agg)); LongTerms.Bucket bucket = new LongTerms.Bucket(1, 1, internalAggregations, false, 0, DocValueFormat.RAW); buckets.add(bucket); @@ -96,7 +96,7 @@ public void testResolveToBucketCount() { AggregationPath path = AggregationPath.parse("_bucket_count"); List buckets = new ArrayList<>(); InternalAggregation agg = new InternalAvg("the_avg", 2, 1, DocValueFormat.RAW, Collections.emptyMap()); - InternalAggregations internalAggregations = new InternalAggregations(Collections.singletonList(agg)); + InternalAggregations internalAggregations = InternalAggregations.from(Collections.singletonList(agg)); LongTerms.Bucket bucket = new LongTerms.Bucket(1, 1, internalAggregations, false, 0, DocValueFormat.RAW); buckets.add(bucket); @@ -109,7 +109,7 @@ public void testResolveToCount() { AggregationPath path = AggregationPath.parse("_count"); List buckets = new ArrayList<>(); InternalAggregation agg = new InternalAvg("the_avg", 2, 1, DocValueFormat.RAW, Collections.emptyMap()); - InternalAggregations internalAggregations = new InternalAggregations(Collections.singletonList(agg)); + InternalAggregations internalAggregations = InternalAggregations.from(Collections.singletonList(agg)); LongTerms.Bucket bucket = new LongTerms.Bucket(1, 1, internalAggregations, false, 0, DocValueFormat.RAW); buckets.add(bucket); @@ -122,7 +122,7 @@ public void testResolveToKey() { AggregationPath path = AggregationPath.parse("_key"); List buckets = new ArrayList<>(); InternalAggregation agg = new InternalAvg("the_avg", 2, 1, DocValueFormat.RAW, Collections.emptyMap()); - InternalAggregations internalAggregations = new InternalAggregations(Collections.singletonList(agg)); + InternalAggregations internalAggregations = InternalAggregations.from(Collections.singletonList(agg)); LongTerms.Bucket bucket = new LongTerms.Bucket(19, 1, internalAggregations, false, 0, DocValueFormat.RAW); buckets.add(bucket); @@ -136,14 +136,14 @@ public void testResolveToSpecificBucket() { List buckets = new ArrayList<>(); InternalAggregation agg = new InternalAvg("the_avg", 2, 1, DocValueFormat.RAW, Collections.emptyMap()); - InternalAggregations internalStringAggs = new InternalAggregations(Collections.singletonList(agg)); + InternalAggregations internalStringAggs = InternalAggregations.from(Collections.singletonList(agg)); List stringBuckets = Collections.singletonList(new StringTerms.Bucket( new BytesRef("foo".getBytes(StandardCharsets.UTF_8), 0, "foo".getBytes(StandardCharsets.UTF_8).length), 1, internalStringAggs, false, 0, DocValueFormat.RAW)); InternalTerms termsAgg = new StringTerms("string_terms", BucketOrder.count(false), 1, 0, Collections.emptyMap(), DocValueFormat.RAW, 1, false, 0, stringBuckets, 0); - InternalAggregations internalAggregations = new InternalAggregations(Collections.singletonList(termsAgg)); + InternalAggregations internalAggregations = InternalAggregations.from(Collections.singletonList(termsAgg)); LongTerms.Bucket bucket = new LongTerms.Bucket(19, 1, internalAggregations, false, 0, DocValueFormat.RAW); buckets.add(bucket); @@ -156,14 +156,14 @@ public void testResolveToMissingSpecificBucket() { List buckets = new ArrayList<>(); InternalAggregation agg = new InternalAvg("the_avg", 2, 1, DocValueFormat.RAW, Collections.emptyMap()); - InternalAggregations internalStringAggs = new InternalAggregations(Collections.singletonList(agg)); + InternalAggregations internalStringAggs = InternalAggregations.from(Collections.singletonList(agg)); List stringBuckets = Collections.singletonList(new StringTerms.Bucket( new BytesRef("foo".getBytes(StandardCharsets.UTF_8), 0, "foo".getBytes(StandardCharsets.UTF_8).length), 1, internalStringAggs, false, 0, DocValueFormat.RAW)); InternalTerms termsAgg = new StringTerms("string_terms", BucketOrder.count(false), 1, 0, Collections.emptyMap(), DocValueFormat.RAW, 1, false, 0, stringBuckets, 0); - InternalAggregations internalAggregations = new InternalAggregations(Collections.singletonList(termsAgg)); + InternalAggregations internalAggregations = InternalAggregations.from(Collections.singletonList(termsAgg)); LongTerms.Bucket bucket = new LongTerms.Bucket(19, 1, internalAggregations, false, 0, DocValueFormat.RAW); buckets.add(bucket); diff --git a/server/src/test/java/org/elasticsearch/search/aggregations/bucket/filter/InternalFilterTests.java b/server/src/test/java/org/elasticsearch/search/aggregations/bucket/filter/InternalFilterTests.java index c90234e04a2e7..b7f5ae06ca503 100644 --- a/server/src/test/java/org/elasticsearch/search/aggregations/bucket/filter/InternalFilterTests.java +++ b/server/src/test/java/org/elasticsearch/search/aggregations/bucket/filter/InternalFilterTests.java @@ -27,6 +27,7 @@ import org.elasticsearch.search.aggregations.pipeline.PipelineAggregator; import org.elasticsearch.search.aggregations.pipeline.PipelineAggregator.PipelineTree; +import java.util.Collections; import java.util.List; import java.util.Map; @@ -65,7 +66,7 @@ public void testReducePipelinesReducesBucketPipelines() { InternalFilter dummy = createTestInstance(); InternalFilter inner = createTestInstance(); - InternalAggregations sub = new InternalAggregations(singletonList(inner)); + InternalAggregations sub = InternalAggregations.from(Collections.singletonList(inner)); InternalFilter test = createTestInstance("test", randomNonNegativeLong(), sub, emptyMap()); PipelineAggregator mockPipeline = new PipelineAggregator(null, null, null) { @Override diff --git a/server/src/test/java/org/elasticsearch/search/aggregations/bucket/filter/InternalFiltersTests.java b/server/src/test/java/org/elasticsearch/search/aggregations/bucket/filter/InternalFiltersTests.java index caf7d25a1df75..0282e8f95591b 100644 --- a/server/src/test/java/org/elasticsearch/search/aggregations/bucket/filter/InternalFiltersTests.java +++ b/server/src/test/java/org/elasticsearch/search/aggregations/bucket/filter/InternalFiltersTests.java @@ -29,6 +29,7 @@ import org.elasticsearch.test.InternalMultiBucketAggregationTestCase; import java.util.ArrayList; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -132,7 +133,7 @@ public void testReducePipelinesReducesBucketPipelines() { InternalFilters dummy = createTestInstance(); InternalFilters inner = createTestInstance(); - InternalAggregations sub = new InternalAggregations(singletonList(inner)); + InternalAggregations sub = InternalAggregations.from(Collections.singletonList(inner)); InternalFilters test = createTestInstance("test", emptyMap(), sub); PipelineAggregator mockPipeline = new PipelineAggregator(null, null, null) { @Override diff --git a/server/src/test/java/org/elasticsearch/search/aggregations/bucket/histogram/InternalAutoDateHistogramTests.java b/server/src/test/java/org/elasticsearch/search/aggregations/bucket/histogram/InternalAutoDateHistogramTests.java index 4d7e9b731a56e..f487b783ae159 100644 --- a/server/src/test/java/org/elasticsearch/search/aggregations/bucket/histogram/InternalAutoDateHistogramTests.java +++ b/server/src/test/java/org/elasticsearch/search/aggregations/bucket/histogram/InternalAutoDateHistogramTests.java @@ -37,7 +37,6 @@ import java.time.ZonedDateTime; import java.util.ArrayList; import java.util.Arrays; -import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -73,8 +72,7 @@ protected InternalAutoDateHistogram createTestInstance( long key = startingDate + (intervalMillis * i); buckets.add(i, new InternalAutoDateHistogram.Bucket(key, randomIntBetween(1, 100), format, aggregations)); } - InternalAggregations subAggregations = new InternalAggregations(Collections.emptyList()); - BucketInfo bucketInfo = new BucketInfo(roundingInfos, roundingIndex, subAggregations); + BucketInfo bucketInfo = new BucketInfo(roundingInfos, roundingIndex, InternalAggregations.EMPTY); return new InternalAutoDateHistogram(name, buckets, targetBuckets, bucketInfo, format, metadata, 1); } @@ -360,7 +358,7 @@ private static class ReduceTestBuilder { ReduceTestBuilder bucket(String key, long docCount) { buckets.add(new InternalAutoDateHistogram.Bucket( - utcMillis(key), docCount, FORMAT, new InternalAggregations(emptyList()))); + utcMillis(key), docCount, FORMAT, InternalAggregations.EMPTY)); return this; } @@ -376,7 +374,7 @@ ReduceTestBuilder finishShardResult(String whichRounding, int innerInterval) { assertThat("rounding [" + whichRounding + "] should be in " + Arrays.toString(roundings), roundingIdx, greaterThan(-1)); assertTrue(Arrays.toString(roundings[roundingIdx].innerIntervals) + " must contain " + innerInterval, Arrays.binarySearch(roundings[roundingIdx].innerIntervals, innerInterval) >= 0); - BucketInfo bucketInfo = new BucketInfo(roundings, roundingIdx, new InternalAggregations(emptyList())); + BucketInfo bucketInfo = new BucketInfo(roundings, roundingIdx, InternalAggregations.EMPTY); results.add(new InternalAutoDateHistogram("test", new ArrayList<>(buckets), targetBuckets, bucketInfo, FORMAT, emptyMap(), innerInterval)); buckets.clear(); diff --git a/test/framework/src/main/java/org/elasticsearch/search/aggregations/InternalSingleBucketAggregationTestCase.java b/test/framework/src/main/java/org/elasticsearch/search/aggregations/InternalSingleBucketAggregationTestCase.java index 2901d80812272..4af7de4f61d08 100644 --- a/test/framework/src/main/java/org/elasticsearch/search/aggregations/InternalSingleBucketAggregationTestCase.java +++ b/test/framework/src/main/java/org/elasticsearch/search/aggregations/InternalSingleBucketAggregationTestCase.java @@ -62,7 +62,7 @@ public void setUp() throws Exception { if (hasInternalMin) { aggs.add(new InternalMin("min", randomDouble(), randomNumericDocValueFormat(), emptyMap())); } - return new InternalAggregations(aggs); + return InternalAggregations.from(aggs); }; } @@ -93,7 +93,7 @@ protected T mutateInstance(T instance) { List aggs = new ArrayList<>(); aggs.add(new InternalMax("new_max", randomDouble(), randomNumericDocValueFormat(), emptyMap())); aggs.add(new InternalMin("new_min", randomDouble(), randomNumericDocValueFormat(), emptyMap())); - aggregations = new InternalAggregations(aggs); + aggregations = InternalAggregations.from(aggs); break; case 3: default: diff --git a/test/framework/src/main/java/org/elasticsearch/test/InternalMultiBucketAggregationTestCase.java b/test/framework/src/main/java/org/elasticsearch/test/InternalMultiBucketAggregationTestCase.java index ec8204b362aed..44358c666fadb 100644 --- a/test/framework/src/main/java/org/elasticsearch/test/InternalMultiBucketAggregationTestCase.java +++ b/test/framework/src/main/java/org/elasticsearch/test/InternalMultiBucketAggregationTestCase.java @@ -78,7 +78,7 @@ public void setUp() throws Exception { for (int i = 0; i < numAggregations; i++) { aggs.add(createTestInstance(randomAlphaOfLength(5), emptyMap(), InternalAggregations.EMPTY)); } - return new InternalAggregations(aggs); + return InternalAggregations.from(aggs); }; } } diff --git a/x-pack/plugin/analytics/src/main/java/org/elasticsearch/xpack/analytics/cumulativecardinality/CumulativeCardinalityPipelineAggregator.java b/x-pack/plugin/analytics/src/main/java/org/elasticsearch/xpack/analytics/cumulativecardinality/CumulativeCardinalityPipelineAggregator.java index 7e8ad20b0f102..7e18f74658c51 100644 --- a/x-pack/plugin/analytics/src/main/java/org/elasticsearch/xpack/analytics/cumulativecardinality/CumulativeCardinalityPipelineAggregator.java +++ b/x-pack/plugin/analytics/src/main/java/org/elasticsearch/xpack/analytics/cumulativecardinality/CumulativeCardinalityPipelineAggregator.java @@ -83,7 +83,7 @@ public InternalAggregation reduce(InternalAggregation aggregation, ReduceContext .map((p) -> (InternalAggregation) p) .collect(Collectors.toList()); aggs.add(new InternalSimpleLongValue(name(), cardinality, formatter, metadata())); - Bucket newBucket = factory.createBucket(factory.getKey(bucket), bucket.getDocCount(), new InternalAggregations(aggs)); + Bucket newBucket = factory.createBucket(factory.getKey(bucket), bucket.getDocCount(), InternalAggregations.from(aggs)); newBuckets.add(newBucket); } return factory.createAggregation(newBuckets); diff --git a/x-pack/plugin/analytics/src/main/java/org/elasticsearch/xpack/analytics/movingPercentiles/MovingPercentilesPipelineAggregator.java b/x-pack/plugin/analytics/src/main/java/org/elasticsearch/xpack/analytics/movingPercentiles/MovingPercentilesPipelineAggregator.java index 3d96dbd1c151b..8eecd627e17a1 100644 --- a/x-pack/plugin/analytics/src/main/java/org/elasticsearch/xpack/analytics/movingPercentiles/MovingPercentilesPipelineAggregator.java +++ b/x-pack/plugin/analytics/src/main/java/org/elasticsearch/xpack/analytics/movingPercentiles/MovingPercentilesPipelineAggregator.java @@ -106,7 +106,7 @@ private void reduceTDigest(List (InternalAggregation) p) .collect(Collectors.toList()); aggs.add(new InternalTDigestPercentiles(name(), config.keys, state, config.keyed, config.formatter, metadata())); - newBucket = factory.createBucket(factory.getKey(bucket), bucket.getDocCount(), new InternalAggregations(aggs)); + newBucket = factory.createBucket(factory.getKey(bucket), bucket.getDocCount(), InternalAggregations.from(aggs)); } newBuckets.add(newBucket); index++; @@ -152,7 +152,7 @@ private void reduceHDR(List (InternalAggregation) p) .collect(Collectors.toList()); aggs.add(new InternalHDRPercentiles(name(), config.keys, state, config.keyed, config.formatter, metadata())); - newBucket = factory.createBucket(factory.getKey(bucket), bucket.getDocCount(), new InternalAggregations(aggs)); + newBucket = factory.createBucket(factory.getKey(bucket), bucket.getDocCount(), InternalAggregations.from(aggs)); } newBuckets.add(newBucket); index++; diff --git a/x-pack/plugin/analytics/src/main/java/org/elasticsearch/xpack/analytics/normalize/NormalizePipelineAggregator.java b/x-pack/plugin/analytics/src/main/java/org/elasticsearch/xpack/analytics/normalize/NormalizePipelineAggregator.java index cf18c49e7c81b..e831700f1335e 100644 --- a/x-pack/plugin/analytics/src/main/java/org/elasticsearch/xpack/analytics/normalize/NormalizePipelineAggregator.java +++ b/x-pack/plugin/analytics/src/main/java/org/elasticsearch/xpack/analytics/normalize/NormalizePipelineAggregator.java @@ -67,7 +67,7 @@ public InternalAggregation reduce(InternalAggregation aggregation, ReduceContext .map((p) -> (InternalAggregation) p) .collect(Collectors.toList()); aggs.add(new InternalSimpleValue(name(), normalizedBucketValue, formatter, metadata())); - InternalMultiBucketAggregation.InternalBucket newBucket = originalAgg.createBucket(new InternalAggregations(aggs), bucket); + InternalMultiBucketAggregation.InternalBucket newBucket = originalAgg.createBucket(InternalAggregations.from(aggs), bucket); newBuckets.add(newBucket); } diff --git a/x-pack/plugin/async-search/src/test/java/org/elasticsearch/xpack/search/AsyncSearchTaskTests.java b/x-pack/plugin/async-search/src/test/java/org/elasticsearch/xpack/search/AsyncSearchTaskTests.java index 2bec519882632..cc11e7613a045 100644 --- a/x-pack/plugin/async-search/src/test/java/org/elasticsearch/xpack/search/AsyncSearchTaskTests.java +++ b/x-pack/plugin/async-search/src/test/java/org/elasticsearch/xpack/search/AsyncSearchTaskTests.java @@ -146,12 +146,12 @@ public void testGetResponseFailureDuringReduction() throws InterruptedException AsyncSearchTask task = createAsyncSearchTask(); task.getSearchProgressActionListener().onListShards(Collections.emptyList(), Collections.emptyList(), SearchResponse.Clusters.EMPTY, false); - InternalAggregations aggs = new InternalAggregations(Collections.singletonList(new StringTerms("name", BucketOrder.key(true), 1, 1, + InternalAggregations aggs = InternalAggregations.from(Collections.singletonList(new StringTerms("name", BucketOrder.key(true), 1, 1, Collections.emptyMap(), DocValueFormat.RAW, 1, false, 1, Collections.emptyList(), 0))); //providing an empty named writeable registry will make the expansion fail, hence the delayed reduction will fail too //causing an exception when executing getResponse as part of the completion listener callback DelayableWriteable.Serialized serializedAggs = DelayableWriteable.referencing(aggs) - .asSerialized(InternalAggregations::new, new NamedWriteableRegistry(Collections.emptyList())); + .asSerialized(InternalAggregations::readFrom, new NamedWriteableRegistry(Collections.emptyList())); task.getSearchProgressActionListener().onPartialReduce(Collections.emptyList(), new TotalHits(0, TotalHits.Relation.EQUAL_TO), serializedAggs, 1); AtomicReference response = new AtomicReference<>(); @@ -184,12 +184,12 @@ public void testWithFailureAndGetResponseFailureDuringReduction() throws Interru AsyncSearchTask task = createAsyncSearchTask(); task.getSearchProgressActionListener().onListShards(Collections.emptyList(), Collections.emptyList(), SearchResponse.Clusters.EMPTY, false); - InternalAggregations aggs = new InternalAggregations(Collections.singletonList(new StringTerms("name", BucketOrder.key(true), 1, 1, + InternalAggregations aggs = InternalAggregations.from(Collections.singletonList(new StringTerms("name", BucketOrder.key(true), 1, 1, Collections.emptyMap(), DocValueFormat.RAW, 1, false, 1, Collections.emptyList(), 0))); //providing an empty named writeable registry will make the expansion fail, hence the delayed reduction will fail too //causing an exception when executing getResponse as part of the completion listener callback DelayableWriteable.Serialized serializedAggs = DelayableWriteable.referencing(aggs) - .asSerialized(InternalAggregations::new, new NamedWriteableRegistry(Collections.emptyList())); + .asSerialized(InternalAggregations::readFrom, new NamedWriteableRegistry(Collections.emptyList())); task.getSearchProgressActionListener().onPartialReduce(Collections.emptyList(), new TotalHits(0, TotalHits.Relation.EQUAL_TO), serializedAggs, 1); task.getSearchProgressActionListener().onFailure(new CircuitBreakingException("boom", CircuitBreaker.Durability.TRANSIENT)); diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/inference/aggs/InferencePipelineAggregator.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/inference/aggs/InferencePipelineAggregator.java index 31ea048cd9e7f..8a0eb04df3447 100644 --- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/inference/aggs/InferencePipelineAggregator.java +++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/inference/aggs/InferencePipelineAggregator.java @@ -107,7 +107,7 @@ public InternalAggregation reduce(InternalAggregation aggregation, InternalAggre InternalInferenceAggregation aggResult = new InternalInferenceAggregation(name(), metadata(), inference); aggs.add(aggResult); - InternalMultiBucketAggregation.InternalBucket newBucket = originalAgg.createBucket(new InternalAggregations(aggs), bucket); + InternalMultiBucketAggregation.InternalBucket newBucket = originalAgg.createBucket(InternalAggregations.from(aggs), bucket); newBuckets.add(newBucket); } diff --git a/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/rollup/RollupResponseTranslator.java b/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/rollup/RollupResponseTranslator.java index ffbbf86cb4866..9f49a4352a7ce 100644 --- a/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/rollup/RollupResponseTranslator.java +++ b/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/rollup/RollupResponseTranslator.java @@ -5,8 +5,6 @@ */ package org.elasticsearch.xpack.rollup; -import org.apache.logging.log4j.LogManager; -import org.apache.logging.log4j.Logger; import org.apache.lucene.util.BytesRef; import org.elasticsearch.ResourceNotFoundException; import org.elasticsearch.action.search.MultiSearchResponse; @@ -41,7 +39,6 @@ import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Arrays; -import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -57,8 +54,6 @@ */ public class RollupResponseTranslator { - private static final Logger logger = LogManager.getLogger(RollupResponseTranslator.class); - /** * Verifies a live-only search response. Essentially just checks for failure then returns * the response since we have no work to do @@ -272,7 +267,7 @@ private static SearchResponse doCombineResponse(SearchResponse liveResponse, Lis // The combination process returns a tree that is identical to the non-rolled // which means we can use aggregation's reduce method to combine, just as if // it was a result from another shard - InternalAggregations currentTree = new InternalAggregations(Collections.emptyList()); + InternalAggregations currentTree = InternalAggregations.EMPTY; InternalAggregation.ReduceContext finalReduceContext = InternalAggregation.ReduceContext.forFinalReduction( reduceContext.bigArrays(), reduceContext.scriptService(), b -> {}, PipelineTree.EMPTY); for (SearchResponse rolledResponse : rolledResponses) { @@ -291,7 +286,7 @@ private static SearchResponse doCombineResponse(SearchResponse liveResponse, Lis // Iteratively merge in each new set of unrolled aggs, so that we can identify/fix overlapping doc_counts // in the next round of unrolling - InternalAggregations finalUnrolledAggs = new InternalAggregations(unrolledAggs); + InternalAggregations finalUnrolledAggs = InternalAggregations.from(unrolledAggs); currentTree = InternalAggregations.reduce(Arrays.asList(currentTree, finalUnrolledAggs), finalReduceContext); } @@ -505,7 +500,7 @@ InternalAggregation unrollMultiBucket(T source, T original, T currentTree, */ private static InternalAggregations unrollSubAggsFromMulti(InternalBucket bucket, InternalBucket original, InternalBucket currentTree) { // Iterate over the subAggs in each bucket - return new InternalAggregations(bucket.getAggregations() + return InternalAggregations.from(bucket.getAggregations() .asList().stream() // Avoid any rollup count metrics, as that's not a true "sub-agg" but rather agg // added by the rollup for accounting purposes (e.g. doc_count) diff --git a/x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/rollup/RollupResponseTranslationTests.java b/x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/rollup/RollupResponseTranslationTests.java index b6a11f4e4bd7e..e4800e49e7b22 100644 --- a/x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/rollup/RollupResponseTranslationTests.java +++ b/x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/rollup/RollupResponseTranslationTests.java @@ -162,11 +162,11 @@ public void testMissingLiveIndex() throws Exception { when(count.getType()).thenReturn(SumAggregationBuilder.NAME); subaggs.add(count); - when(filter.getAggregations()).thenReturn(new InternalAggregations(subaggs)); + when(filter.getAggregations()).thenReturn(InternalAggregations.from(subaggs)); when(filter.getName()).thenReturn("filter_foo"); aggTree.add(filter); - Aggregations mockAggsWithout = new InternalAggregations(aggTree); + Aggregations mockAggsWithout = InternalAggregations.from(aggTree); when(responseWithout.getAggregations()).thenReturn(mockAggsWithout); MultiSearchResponse.Item[] msearch = new MultiSearchResponse.Item[]{ @@ -186,8 +186,7 @@ public void testRolledMissingAggs() throws Exception { SearchResponse responseWithout = mock(SearchResponse.class); when(responseWithout.getTook()).thenReturn(new TimeValue(100)); - Aggregations mockAggsWithout = new InternalAggregations(Collections.emptyList()); - when(responseWithout.getAggregations()).thenReturn(mockAggsWithout); + when(responseWithout.getAggregations()).thenReturn(InternalAggregations.EMPTY); MultiSearchResponse.Item[] msearch = new MultiSearchResponse.Item[]{ new MultiSearchResponse.Item(responseWithout, null)}; @@ -258,11 +257,11 @@ public void testTranslateRollup() throws Exception { when(count.getType()).thenReturn(SumAggregationBuilder.NAME); subaggs.add(count); - when(filter.getAggregations()).thenReturn(new InternalAggregations(subaggs)); + when(filter.getAggregations()).thenReturn(InternalAggregations.from(subaggs)); when(filter.getName()).thenReturn("filter_foo"); aggTree.add(filter); - Aggregations mockAggs = new InternalAggregations(aggTree); + Aggregations mockAggs = InternalAggregations.from(aggTree); when(response.getAggregations()).thenReturn(mockAggs); MultiSearchResponse.Item item = new MultiSearchResponse.Item(response, null); @@ -298,7 +297,7 @@ public void testMissingFilter() { InternalMax protoMax = mock(InternalMax.class); when(protoMax.getName()).thenReturn("foo"); protoAggTree.add(protoMax); - Aggregations protoMockAggs = new InternalAggregations(protoAggTree); + Aggregations protoMockAggs = InternalAggregations.from(protoAggTree); when(protoResponse.getAggregations()).thenReturn(protoMockAggs); MultiSearchResponse.Item unrolledResponse = new MultiSearchResponse.Item(protoResponse, null); @@ -307,7 +306,7 @@ public void testMissingFilter() { InternalMax max = mock(InternalMax.class); when(max.getName()).thenReturn("bizzbuzz"); aggTreeWithoutFilter.add(max); - Aggregations mockAggsWithout = new InternalAggregations(aggTreeWithoutFilter); + Aggregations mockAggsWithout = InternalAggregations.from(aggTreeWithoutFilter); when(responseWithout.getAggregations()).thenReturn(mockAggsWithout); MultiSearchResponse.Item rolledResponse = new MultiSearchResponse.Item(responseWithout, null); @@ -328,7 +327,7 @@ public void testMatchingNameNotFilter() { InternalMax protoMax = mock(InternalMax.class); when(protoMax.getName()).thenReturn("foo"); protoAggTree.add(protoMax); - Aggregations protoMockAggs = new InternalAggregations(protoAggTree); + Aggregations protoMockAggs = InternalAggregations.from(protoAggTree); when(protoResponse.getAggregations()).thenReturn(protoMockAggs); MultiSearchResponse.Item unrolledResponse = new MultiSearchResponse.Item(protoResponse, null); @@ -336,7 +335,7 @@ public void testMatchingNameNotFilter() { List aggTreeWithoutFilter = new ArrayList<>(1); InternalMax max = new InternalMax("filter_foo", 0, DocValueFormat.RAW, null); aggTreeWithoutFilter.add(max); - Aggregations mockAggsWithout = new InternalAggregations(aggTreeWithoutFilter); + Aggregations mockAggsWithout = InternalAggregations.from(aggTreeWithoutFilter); when(responseWithout.getAggregations()).thenReturn(mockAggsWithout); MultiSearchResponse.Item rolledResponse = new MultiSearchResponse.Item(responseWithout, null); @@ -358,7 +357,7 @@ public void testSimpleReduction() throws Exception { List protoAggTree = new ArrayList<>(1); InternalAvg internalAvg = new InternalAvg("foo", 10, 2, DocValueFormat.RAW, null); protoAggTree.add(internalAvg); - Aggregations protoMockAggs = new InternalAggregations(protoAggTree); + Aggregations protoMockAggs = InternalAggregations.from(protoAggTree); when(protoResponse.getAggregations()).thenReturn(protoMockAggs); MultiSearchResponse.Item unrolledResponse = new MultiSearchResponse.Item(protoResponse, null); @@ -386,11 +385,11 @@ public void testSimpleReduction() throws Exception { when(count.getType()).thenReturn(SumAggregationBuilder.NAME); subaggs.add(count); - when(filter.getAggregations()).thenReturn(new InternalAggregations(subaggs)); + when(filter.getAggregations()).thenReturn(InternalAggregations.from(subaggs)); when(filter.getName()).thenReturn("filter_foo"); aggTree.add(filter); - Aggregations mockAggsWithout = new InternalAggregations(aggTree); + Aggregations mockAggsWithout = InternalAggregations.from(aggTree); when(responseWithout.getAggregations()).thenReturn(mockAggsWithout); MultiSearchResponse.Item rolledResponse = new MultiSearchResponse.Item(responseWithout, null); @@ -495,11 +494,11 @@ public void testMismatch() throws IOException { // TODO SearchResponse.Clusters is not public, using null for now. Should fix upstream. MultiSearchResponse.Item unrolledItem = new MultiSearchResponse.Item(new SearchResponse( new InternalSearchResponse(null, - new InternalAggregations(Collections.singletonList(responses.get(0))), null, null, false, false, 1), + InternalAggregations.from(Collections.singletonList(responses.get(0))), null, null, false, false, 1), null, 1, 1, 0, 10, null, null), null); MultiSearchResponse.Item rolledItem = new MultiSearchResponse.Item(new SearchResponse( new InternalSearchResponse(null, - new InternalAggregations(Collections.singletonList(responses.get(1))), null, null, false, false, 1), + InternalAggregations.from(Collections.singletonList(responses.get(1))), null, null, false, false, 1), null, 1, 1, 0, 10, null, null), null); MultiSearchResponse.Item[] msearch = new MultiSearchResponse.Item[]{unrolledItem, rolledItem}; diff --git a/x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/rollup/action/SearchActionTests.java b/x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/rollup/action/SearchActionTests.java index 0ad7d13d4fb13..500eec809f04c 100644 --- a/x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/rollup/action/SearchActionTests.java +++ b/x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/rollup/action/SearchActionTests.java @@ -657,11 +657,11 @@ public void testRollupOnly() throws Exception { when(count.getType()).thenReturn(SumAggregationBuilder.NAME); subaggs.add(count); - when(filter.getAggregations()).thenReturn(new InternalAggregations(subaggs)); + when(filter.getAggregations()).thenReturn(InternalAggregations.from(subaggs)); when(filter.getName()).thenReturn("filter_foo"); aggTree.add(filter); - Aggregations mockAggs = new InternalAggregations(aggTree); + Aggregations mockAggs = InternalAggregations.from(aggTree); when(response.getAggregations()).thenReturn(mockAggs); MultiSearchResponse.Item item = new MultiSearchResponse.Item(response, null); MultiSearchResponse msearchResponse = new MultiSearchResponse(new MultiSearchResponse.Item[]{item}, 1); @@ -747,7 +747,7 @@ public void testBoth() throws Exception { List protoAggTree = new ArrayList<>(1); InternalAvg internalAvg = new InternalAvg("foo", 10, 2, DocValueFormat.RAW, null); protoAggTree.add(internalAvg); - Aggregations protoMockAggs = new InternalAggregations(protoAggTree); + Aggregations protoMockAggs = InternalAggregations.from(protoAggTree); when(protoResponse.getAggregations()).thenReturn(protoMockAggs); MultiSearchResponse.Item unrolledResponse = new MultiSearchResponse.Item(protoResponse, null); @@ -775,11 +775,11 @@ public void testBoth() throws Exception { when(count.getType()).thenReturn(SumAggregationBuilder.NAME); subaggs.add(count); - when(filter.getAggregations()).thenReturn(new InternalAggregations(subaggs)); + when(filter.getAggregations()).thenReturn(InternalAggregations.from(subaggs)); when(filter.getName()).thenReturn("filter_foo"); aggTree.add(filter); - Aggregations mockAggsWithout = new InternalAggregations(aggTree); + Aggregations mockAggsWithout = InternalAggregations.from(aggTree); when(responseWithout.getAggregations()).thenReturn(mockAggsWithout); MultiSearchResponse.Item rolledResponse = new MultiSearchResponse.Item(responseWithout, null); diff --git a/x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/rollup/job/RollupIndexerStateTests.java b/x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/rollup/job/RollupIndexerStateTests.java index 51b9bb4e315be..5be2fe83c77a8 100644 --- a/x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/rollup/job/RollupIndexerStateTests.java +++ b/x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/rollup/job/RollupIndexerStateTests.java @@ -634,7 +634,7 @@ public long getDocCount() { @Override public Aggregations getAggregations() { - return new InternalAggregations(Collections.emptyList()); + return InternalAggregations.EMPTY; } @Override @@ -747,7 +747,7 @@ public long getDocCount() { @Override public Aggregations getAggregations() { - return new InternalAggregations(Collections.emptyList()); + return InternalAggregations.EMPTY; } @Override @@ -904,7 +904,7 @@ public long getDocCount() { @Override public Aggregations getAggregations() { - return new InternalAggregations(Collections.emptyList()); + return InternalAggregations.EMPTY; } @Override