Move InternalAggregations to Writeable (#41841)

javanna · web-flow · commit 3eb85751d2c6 · 2019-05-07T10:54:47.000+02:00
Relates to #34389
diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/Aggregations.java b/server/src/main/java/org/elasticsearch/search/aggregations/Aggregations.java
@@ -34,6 +34,7 @@
 import java.util.Map;
 import java.util.Objects;
 
+import static java.util.Collections.emptyMap;
 import static java.util.Collections.unmodifiableMap;
 import static org.elasticsearch.common.xcontent.XContentParserUtils.parseTypedKeysObject;
 
@@ -44,14 +45,14 @@ public class Aggregations implements Iterable<Aggregation>, ToXContentFragment {
 
     public static final String AGGREGATIONS_FIELD = "aggregations";
 
-    protected List<? extends Aggregation> aggregations = Collections.emptyList();
-    protected Map<String, Aggregation> aggregationsAsMap;
-
-    protected Aggregations() {
-    }
+    protected final List<? extends Aggregation> aggregations;
+    private Map<String, Aggregation> aggregationsAsMap;
 
     public Aggregations(List<? extends Aggregation> aggregations) {
         this.aggregations = aggregations;
+        if (aggregations.isEmpty()) {
+            aggregationsAsMap = emptyMap();
+        }
     }
 
     /**
diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/InternalAggregations.java b/server/src/main/java/org/elasticsearch/search/aggregations/InternalAggregations.java
@@ -20,7 +20,7 @@
 
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
-import org.elasticsearch.common.io.stream.Streamable;
+import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.search.aggregations.InternalAggregation.ReduceContext;
 import org.elasticsearch.search.aggregations.pipeline.PipelineAggregator;
 import org.elasticsearch.search.aggregations.pipeline.SiblingPipelineAggregator;
@@ -34,14 +34,13 @@
 import java.util.Map;
 import java.util.Objects;
 
-import static java.util.Collections.emptyMap;
-
 /**
  * An internal implementation of {@link Aggregations}.
  */
-public final class InternalAggregations extends Aggregations implements Streamable {
+public final class InternalAggregations extends Aggregations implements Writeable {
+
+    public static final InternalAggregations EMPTY = new InternalAggregations(Collections.emptyList());
 
-    public static final InternalAggregations EMPTY = new InternalAggregations();
     private static final Comparator<InternalAggregation> INTERNAL_AGG_COMPARATOR = (agg1, agg2) -> {
         if (agg1.isMapped() == agg2.isMapped()) {
             return 0;
@@ -52,16 +51,14 @@ public final class InternalAggregations extends Aggregations implements Streamab
         }
     };
 
-    private List<SiblingPipelineAggregator> topLevelPipelineAggregators = Collections.emptyList();
-
-    private InternalAggregations() {
-    }
+    private final List<SiblingPipelineAggregator> topLevelPipelineAggregators;
 
     /**
      * Constructs a new aggregation.
      */
     public InternalAggregations(List<InternalAggregation> aggregations) {
         super(aggregations);
+        this.topLevelPipelineAggregators = Collections.emptyList();
     }
 
     /**
@@ -72,6 +69,19 @@ public InternalAggregations(List<InternalAggregation> aggregations, List<Sibling
         this.topLevelPipelineAggregators = Objects.requireNonNull(topLevelPipelineAggregators);
     }
 
+    public InternalAggregations(StreamInput in) throws IOException {
+        super(in.readList(stream -> in.readNamedWriteable(InternalAggregation.class)));
+        this.topLevelPipelineAggregators = in.readList(
+            stream -> (SiblingPipelineAggregator)in.readNamedWriteable(PipelineAggregator.class));
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeNamedWriteableList((List<InternalAggregation>)aggregations);
+        out.writeNamedWriteableList(topLevelPipelineAggregators);
+    }
+
     /**
      * Returns the top-level pipeline aggregators.
      * Note that top-level pipeline aggregators become normal aggregation once the final reduction has been performed, after which they
@@ -86,8 +96,7 @@ public List<SiblingPipelineAggregator> getTopLevelPipelineAggregators() {
      * {@link InternalAggregations} object found in the list.
      * Note that top-level pipeline aggregators are reduced only as part of the final reduction phase, otherwise they are left untouched.
      */
-    public static InternalAggregations reduce(List<InternalAggregations> aggregationsList,
-                                              ReduceContext context) {
+    public static InternalAggregations reduce(List<InternalAggregations> aggregationsList, ReduceContext context) {
         if (aggregationsList.isEmpty()) {
             return null;
         }
@@ -123,27 +132,4 @@ public static InternalAggregations reduce(List<InternalAggregations> aggregation
         }
         return new InternalAggregations(reducedAggregations, topLevelPipelineAggregators);
     }
-
-    public static InternalAggregations readAggregations(StreamInput in) throws IOException {
-        InternalAggregations result = new InternalAggregations();
-        result.readFrom(in);
-        return result;
-    }
-
-    @Override
-    public void readFrom(StreamInput in) throws IOException {
-        aggregations = in.readList(stream -> in.readNamedWriteable(InternalAggregation.class));
-        if (aggregations.isEmpty()) {
-            aggregationsAsMap = emptyMap();
-        }
-        this.topLevelPipelineAggregators = in.readList(
-                stream -> (SiblingPipelineAggregator)in.readNamedWriteable(PipelineAggregator.class));
-    }
-
-    @Override
-    @SuppressWarnings("unchecked")
-    public void writeTo(StreamOutput out) throws IOException {
-        out.writeNamedWriteableList((List<InternalAggregation>)aggregations);
-        out.writeNamedWriteableList(topLevelPipelineAggregators);
-    }
 }
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
@@ -60,7 +60,7 @@ protected InternalSingleBucketAggregation(String name, long docCount, InternalAg
     protected InternalSingleBucketAggregation(StreamInput in) throws IOException {
         super(in);
         docCount = in.readVLong();
-        aggregations = InternalAggregations.readAggregations(in);
+        aggregations = new InternalAggregations(in);
     }
 
     @Override
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
@@ -58,7 +58,7 @@ public InternalBucket(String key, long docCount, InternalAggregations aggregatio
         public InternalBucket(StreamInput in) throws IOException {
             key = in.readOptionalString();
             docCount = in.readVLong();
-            aggregations = InternalAggregations.readAggregations(in);
+            aggregations = new InternalAggregations(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
@@ -237,7 +237,7 @@ public static class InternalBucket extends InternalMultiBucketAggregation.Intern
         InternalBucket(StreamInput in, List<String> sourceNames, List<DocValueFormat> formats, int[] reverseMuls) throws IOException {
             this.key = new CompositeKey(in);
             this.docCount = in.readVLong();
-            this.aggregations = InternalAggregations.readAggregations(in);
+            this.aggregations = new InternalAggregations(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
@@ -57,7 +57,7 @@ public InternalBucket(StreamInput in, boolean keyed) throws IOException {
             this.keyed = keyed;
             key = in.readOptionalString();
             docCount = in.readVLong();
-            aggregations = InternalAggregations.readAggregations(in);
+            aggregations = new InternalAggregations(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
@@ -51,7 +51,7 @@ public InternalGeoGridBucket(long hashAsLong, long docCount, InternalAggregation
     public InternalGeoGridBucket(StreamInput in) throws IOException {
         hashAsLong = in.readLong();
         docCount = in.readVLong();
-        aggregations = InternalAggregations.readAggregations(in);
+        aggregations = new InternalAggregations(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
@@ -73,7 +73,7 @@ public Bucket(StreamInput in, DocValueFormat format) throws IOException {
             this.format = format;
             key = in.readLong();
             docCount = in.readVLong();
-            aggregations = InternalAggregations.readAggregations(in);
+            aggregations = new InternalAggregations(in);
         }
 
         @Override
@@ -175,7 +175,7 @@ static class BucketInfo {
                 roundingInfos[i] = new RoundingInfo(in);
             }
             roundingIdx = in.readVInt();
-            emptySubAggregations = InternalAggregations.readAggregations(in);
+            emptySubAggregations = new InternalAggregations(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
@@ -77,7 +77,7 @@ public Bucket(StreamInput in, boolean keyed, DocValueFormat format) throws IOExc
             this.keyed = keyed;
             key = in.readLong();
             docCount = in.readVLong();
-            aggregations = InternalAggregations.readAggregations(in);
+            aggregations = new InternalAggregations(in);
         }
 
         @Override
@@ -186,7 +186,7 @@ static class EmptyBucketInfo {
 
         EmptyBucketInfo(StreamInput in) throws IOException {
             rounding = Rounding.read(in);
-            subAggregations = InternalAggregations.readAggregations(in);
+            subAggregations = new InternalAggregations(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
@@ -73,7 +73,7 @@ public Bucket(StreamInput in, boolean keyed, DocValueFormat format) throws IOExc
             this.keyed = keyed;
             key = in.readDouble();
             docCount = in.readVLong();
-            aggregations = InternalAggregations.readAggregations(in);
+            aggregations = new InternalAggregations(in);
         }
 
         @Override
@@ -178,7 +178,7 @@ static class EmptyBucketInfo {
         }
 
         EmptyBucketInfo(StreamInput in) throws IOException {
-            this(in.readDouble(), in.readDouble(), in.readDouble(), in.readDouble(), InternalAggregations.readAggregations(in));
+            this(in.readDouble(), in.readDouble(), in.readDouble(), in.readDouble(), new InternalAggregations(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
@@ -77,7 +77,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 = InternalAggregations.readAggregations(in);
+            InternalAggregations aggregations = new InternalAggregations(in);
 
             return new Bucket(format, keyed, key, from, to, docCount, aggregations);
         }
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
@@ -256,7 +256,7 @@ public InternalRange(StreamInput in) throws IOException {
         for (int i = 0; i < size; i++) {
             String key = in.readString();
             ranges.add(getFactory().createBucket(key, in.readDouble(), in.readDouble(), in.readVLong(),
-                    InternalAggregations.readAggregations(in), keyed, format));
+                    new InternalAggregations(in), keyed, format));
         }
         this.ranges = ranges;
     }
diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/significant/SignificantLongTerms.java b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/significant/SignificantLongTerms.java
@@ -59,7 +59,7 @@ static class Bucket extends InternalSignificantTerms.Bucket<Bucket> {
             supersetDf = in.readVLong();
             term = in.readLong();
             score = in.readDouble();
-            aggregations = InternalAggregations.readAggregations(in);
+            aggregations = new InternalAggregations(in);
         }
 
         @Override
diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/bucket/significant/SignificantStringTerms.java b/server/src/main/java/org/elasticsearch/search/aggregations/bucket/significant/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 = InternalAggregations.readAggregations(in);
+            aggregations = new InternalAggregations(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
@@ -85,7 +85,7 @@ protected Bucket(StreamInput in, DocValueFormat formatter, boolean showDocCountE
             if (showDocCountError) {
                 docCountError = in.readLong();
             }
-            aggregations = InternalAggregations.readAggregations(in);
+            aggregations = new InternalAggregations(in);
         }
 
         @Override
diff --git 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() ? InternalAggregations.readAggregations(in) : null,
+                in.readBoolean() ? new InternalAggregations(in) : null,
                 in.readBoolean() ? new Suggest(in) : null,
                 in.readBoolean(),
                 in.readOptionalBoolean(),
@@ -64,7 +64,7 @@ public InternalSearchResponse(StreamInput in) throws IOException {
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         hits.writeTo(out);
-        out.writeOptionalStreamable((InternalAggregations)aggregations);
+        out.writeOptionalWriteable((InternalAggregations)aggregations);
         out.writeOptionalWriteable(suggest);
         out.writeBoolean(timedOut);
         out.writeOptionalBoolean(terminatedEarly);
diff --git a/server/src/main/java/org/elasticsearch/search/query/QuerySearchResult.java b/server/src/main/java/org/elasticsearch/search/query/QuerySearchResult.java
@@ -284,7 +284,7 @@ public void readFromWithId(long id, StreamInput in) throws IOException {
         }
         setTopDocs(readTopDocs(in));
         if (hasAggs = in.readBoolean()) {
-            aggregations = InternalAggregations.readAggregations(in);
+            aggregations = new InternalAggregations(in);
         }
         if (in.getVersion().before(Version.V_7_2_0)) {
             List<SiblingPipelineAggregator> pipelineAggregators = in.readNamedWriteableList(PipelineAggregator.class).stream()
diff --git a/server/src/test/java/org/elasticsearch/search/aggregations/InternalAggregationsTests.java b/server/src/test/java/org/elasticsearch/search/aggregations/InternalAggregationsTests.java
@@ -132,7 +132,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 = InternalAggregations.readAggregations(in);
+                InternalAggregations deserialized = new InternalAggregations(in);
                 assertEquals(aggregations.aggregations, deserialized.aggregations);
                 if (aggregations.getTopLevelPipelineAggregators() == null) {
                     assertEquals(0, deserialized.getTopLevelPipelineAggregators().size());

Original file line number	Diff line number	Diff line change
`@@ -60,7 +60,7 @@ protected InternalSingleBucketAggregation(String name, long docCount, InternalAg`
`60`	`60`	`protected InternalSingleBucketAggregation(StreamInput in) throws IOException {`
`61`	`61`	`super(in);`
`62`	`62`	`docCount = in.readVLong();`
`63`		`- aggregations = InternalAggregations.readAggregations(in);`
	`63`	`+ aggregations = new InternalAggregations(in);`
`64`	`64`	`}`
`65`	`65`
`66`	`66`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -58,7 +58,7 @@ public InternalBucket(String key, long docCount, InternalAggregations aggregatio`
`58`	`58`	`public InternalBucket(StreamInput in) throws IOException {`
`59`	`59`	`key = in.readOptionalString();`
`60`	`60`	`docCount = in.readVLong();`
`61`		`- aggregations = InternalAggregations.readAggregations(in);`
	`61`	`+ aggregations = new InternalAggregations(in);`
`62`	`62`	`}`
`63`	`63`
`64`	`64`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -57,7 +57,7 @@ public InternalBucket(StreamInput in, boolean keyed) throws IOException {`
`57`	`57`	`this.keyed = keyed;`
`58`	`58`	`key = in.readOptionalString();`
`59`	`59`	`docCount = in.readVLong();`
`60`		`- aggregations = InternalAggregations.readAggregations(in);`
	`60`	`+ aggregations = new InternalAggregations(in);`
`61`	`61`	`}`
`62`	`62`
`63`	`63`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -51,7 +51,7 @@ public InternalGeoGridBucket(long hashAsLong, long docCount, InternalAggregation`
`51`	`51`	`public InternalGeoGridBucket(StreamInput in) throws IOException {`
`52`	`52`	`hashAsLong = in.readLong();`
`53`	`53`	`docCount = in.readVLong();`
`54`		`- aggregations = InternalAggregations.readAggregations(in);`
	`54`	`+ aggregations = new InternalAggregations(in);`
`55`	`55`	`}`
`56`	`56`
`57`	`57`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -73,7 +73,7 @@ public Bucket(StreamInput in, DocValueFormat format) throws IOException {`
`73`	`73`	`this.format = format;`
`74`	`74`	`key = in.readLong();`
`75`	`75`	`docCount = in.readVLong();`
`76`		`- aggregations = InternalAggregations.readAggregations(in);`
	`76`	`+ aggregations = new InternalAggregations(in);`
`77`	`77`	`}`
`78`	`78`
`79`	`79`	`@Override`
`@@ -175,7 +175,7 @@ static class BucketInfo {`
`175`	`175`	`roundingInfos[i] = new RoundingInfo(in);`
`176`	`176`	`}`
`177`	`177`	`roundingIdx = in.readVInt();`
`178`		`- emptySubAggregations = InternalAggregations.readAggregations(in);`
	`178`	`+ emptySubAggregations = new InternalAggregations(in);`
`179`	`179`	`}`
`180`	`180`
`181`	`181`	`void writeTo(StreamOutput out) throws IOException {`
Original file line number	Diff line number	Diff line change
`@@ -77,7 +77,7 @@ public Bucket(StreamInput in, boolean keyed, DocValueFormat format) throws IOExc`
`77`	`77`	`this.keyed = keyed;`
`78`	`78`	`key = in.readLong();`
`79`	`79`	`docCount = in.readVLong();`
`80`		`- aggregations = InternalAggregations.readAggregations(in);`
	`80`	`+ aggregations = new InternalAggregations(in);`
`81`	`81`	`}`
`82`	`82`
`83`	`83`	`@Override`
`@@ -186,7 +186,7 @@ static class EmptyBucketInfo {`
`186`	`186`
`187`	`187`	`EmptyBucketInfo(StreamInput in) throws IOException {`
`188`	`188`	`rounding = Rounding.read(in);`
`189`		`- subAggregations = InternalAggregations.readAggregations(in);`
	`189`	`+ subAggregations = new InternalAggregations(in);`
`190`	`190`	`bounds = in.readOptionalWriteable(ExtendedBounds::new);`
`191`	`191`	`}`
`192`	`192`