Aggregations: Added Factory for all MultiBucketAggregations to implement

This allows things like reducers to add aggregations to buckets without needing to know how to construct the aggregation or bucket itself.

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

@@ -25,7 +25,8 @@ import org.elasticsearch.search.aggregations.reducers.Reducer;
 import java.util.List;
 import java.util.Map;
 
-public abstract class InternalMultiBucketAggregation extends InternalAggregation implements MultiBucketsAggregation {
+public abstract class InternalMultiBucketAggregation<A extends InternalMultiBucketAggregation, B extends InternalMultiBucketAggregation.InternalBucket>
+        extends InternalAggregation implements MultiBucketsAggregation {
 
     public InternalMultiBucketAggregation() {
     }
@@ -34,6 +35,28 @@ public abstract class InternalMultiBucketAggregation extends InternalAggregation
         super(name, reducers, metaData);
     }
 
+    /**
+     * Create a new copy of this {@link Aggregation} with the same settings as
+     * this {@link Aggregation} and contains the provided buckets.
+     * 
+     * @param buckets
+     *            the buckets to use in the new {@link Aggregation}
+     * @return the new {@link Aggregation}
+     */
+    public abstract A create(List<B> buckets);
+
+    /**
+     * Create a new {@link InternalBucket} using the provided prototype bucket
+     * and aggregations.
+     * 
+     * @param aggregations
+     *            the aggregations for the new bucket
+     * @param prototype
+     *            the bucket to use as a prototype
+     * @return the new bucket
+     */
+    public abstract B createBucket(InternalAggregations aggregations, B prototype);
+
     @Override
     public Object getProperty(List<String> path) {
         if (path.isEmpty()) {
@@ -75,4 +98,13 @@ public abstract class InternalMultiBucketAggregation extends InternalAggregation
             return aggregation.getProperty(path.subList(1, path.size()));
         }
     }
+
+    public static abstract class Factory<A extends InternalMultiBucketAggregation, B extends InternalMultiBucketAggregation.InternalBucket> {
+
+        public abstract String type();
+
+        public abstract A create(List<B> buckets, A prototype);
+
+        public abstract B createBucket(InternalAggregations aggregations, B prototype);
+    }
 }

src/main/java/org/elasticsearch/search/aggregations/bucket/filters/InternalFilters.java

@@ -29,6 +29,7 @@ import org.elasticsearch.search.aggregations.Aggregations;
 import org.elasticsearch.search.aggregations.InternalAggregation;
 import org.elasticsearch.search.aggregations.InternalAggregations;
 import org.elasticsearch.search.aggregations.InternalMultiBucketAggregation;
+import org.elasticsearch.search.aggregations.InternalMultiBucketAggregation.InternalBucket;
 import org.elasticsearch.search.aggregations.bucket.BucketStreamContext;
 import org.elasticsearch.search.aggregations.bucket.BucketStreams;
 import org.elasticsearch.search.aggregations.reducers.Reducer;
@@ -42,7 +43,7 @@ import java.util.Map;
 /**
  *
  */
-public class InternalFilters extends InternalMultiBucketAggregation implements Filters {
+public class InternalFilters extends InternalMultiBucketAggregation<InternalFilters, InternalFilters.Bucket> implements Filters {
 
     public final static Type TYPE = new Type("filters");
 
@@ -175,6 +176,16 @@ public class InternalFilters extends InternalMultiBucketAggregation implements F
         return TYPE;
     }
 
+    @Override
+    public InternalFilters create(List<Bucket> buckets) {
+        return new InternalFilters(this.name, buckets, this.keyed, this.reducers(), this.metaData);
+    }
+
+    @Override
+    public Bucket createBucket(InternalAggregations aggregations, Bucket prototype) {
+        return new Bucket(prototype.key, prototype.docCount, aggregations, prototype.keyed);
+    }
+
     @Override
     public List<Bucket> getBuckets() {
         return buckets;

src/main/java/org/elasticsearch/search/aggregations/bucket/geogrid/InternalGeoHashGrid.java

@@ -46,7 +46,8 @@ import java.util.Map;
  * All geohashes in a grid are of the same precision and held internally as a single long
  * for efficiency's sake.
  */
-public class InternalGeoHashGrid extends InternalMultiBucketAggregation implements GeoHashGrid {
+public class InternalGeoHashGrid extends InternalMultiBucketAggregation<InternalGeoHashGrid, InternalGeoHashGrid.Bucket> implements
+        GeoHashGrid {
 
     public static final Type TYPE = new Type("geohash_grid", "ghcells");
 
@@ -163,7 +164,6 @@ public class InternalGeoHashGrid extends InternalMultiBucketAggregation implemen
             return builder;
         }
     }
-
     private int requiredSize;
     private Collection<Bucket> buckets;
     protected Map<String, Bucket> bucketMap;
@@ -183,6 +183,16 @@ public class InternalGeoHashGrid extends InternalMultiBucketAggregation implemen
         return TYPE;
     }
 
+    @Override
+    public InternalGeoHashGrid create(List<Bucket> buckets) {
+        return new InternalGeoHashGrid(this.name, this.requiredSize, buckets, this.reducers(), this.metaData);
+    }
+
+    @Override
+    public Bucket createBucket(InternalAggregations aggregations, Bucket prototype) {
+        return new Bucket(prototype.geohashAsLong, prototype.docCount, aggregations);
+    }
+
     @Override
     public List<GeoHashGrid.Bucket> getBuckets() {
         Object o = buckets;

src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/InternalDateHistogram.java

@@ -70,6 +70,11 @@ public class InternalDateHistogram {
             return TYPE.name();
         }
 
+        @Override
+        public InternalDateHistogram.Bucket createBucket(InternalAggregations aggregations, InternalDateHistogram.Bucket prototype) {
+            return new Bucket(prototype.key, prototype.docCount, aggregations, prototype.getKeyed(), prototype.formatter, this);
+        }
+
         @Override
         public InternalDateHistogram.Bucket createBucket(Object key, long docCount, InternalAggregations aggregations, boolean keyed,
                 @Nullable ValueFormatter formatter) {

src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/InternalHistogram.java

@@ -52,7 +52,8 @@ import java.util.Map;
 /**
  * TODO should be renamed to InternalNumericHistogram (see comment on {@link Histogram})?
  */
-public class InternalHistogram<B extends InternalHistogram.Bucket> extends InternalMultiBucketAggregation implements Histogram {
+public class InternalHistogram<B extends InternalHistogram.Bucket> extends InternalMultiBucketAggregation<InternalHistogram, B> implements
+        Histogram {
 
     final static Type TYPE = new Type("histogram", "histo");
 
@@ -233,7 +234,7 @@ public class InternalHistogram<B extends InternalHistogram.Bucket> extends Inter
 
     }
 
-    public static class Factory<B extends InternalHistogram.Bucket> {
+    public static class Factory<B extends InternalHistogram.Bucket> extends InternalMultiBucketAggregation.Factory<InternalHistogram<B>, B> {
 
         protected Factory() {
         }
@@ -248,11 +249,17 @@ public class InternalHistogram<B extends InternalHistogram.Bucket> extends Inter
             return new InternalHistogram<>(name, buckets, order, minDocCount, emptyBucketInfo, formatter, keyed, this, reducers, metaData);
         }
 
-        public InternalHistogram<B> create(String name, List<B> buckets, InternalHistogram prototype) {
-            return new InternalHistogram<>(name, buckets, prototype.order, prototype.minDocCount, prototype.emptyBucketInfo,
+        @Override
+        public InternalHistogram<B> create(List<B> buckets, InternalHistogram<B> prototype) {
+            return new InternalHistogram<>(prototype.name, buckets, prototype.order, prototype.minDocCount, prototype.emptyBucketInfo,
                     prototype.formatter, prototype.keyed, this, prototype.reducers(), prototype.metaData);
         }
 
+        @Override
+        public B createBucket(InternalAggregations aggregations, B prototype) {
+            return (B) new Bucket(prototype.key, prototype.docCount, prototype.getKeyed(), prototype.formatter, this, aggregations);
+        }
+
         public B createBucket(Object key, long docCount, InternalAggregations aggregations, boolean keyed,
                 @Nullable ValueFormatter formatter) {
             if (key instanceof Number) {
@@ -306,6 +313,16 @@ public class InternalHistogram<B extends InternalHistogram.Bucket> extends Inter
         return factory;
     }
 
+    @Override
+    public InternalHistogram<B> create(List<B> buckets) {
+        return getFactory().create(buckets, this);
+    }
+
+    @Override
+    public B createBucket(InternalAggregations aggregations, B prototype) {
+        return getFactory().createBucket(aggregations, prototype);
+    }
+
     private static class IteratorAndCurrent<B> {
 
         private final Iterator<B> iterator;

src/main/java/org/elasticsearch/search/aggregations/bucket/range/InternalRange.java

@@ -43,7 +43,8 @@ import java.util.Map;
 /**
  *
  */
-public class InternalRange<B extends InternalRange.Bucket> extends InternalMultiBucketAggregation implements Range {
+public class InternalRange<B extends InternalRange.Bucket, R extends InternalRange<B, R>> extends InternalMultiBucketAggregation<R, B>
+        implements Range {
 
     static final Factory FACTORY = new Factory();
 
@@ -124,6 +125,14 @@ public class InternalRange<B extends InternalRange.Bucket> extends InternalMulti
             return to;
         }
 
+        public boolean getKeyed() {
+            return keyed;
+        }
+
+        public ValueFormatter getFormatter() {
+            return formatter;
+        }
+
         @Override
         public String getFromAsString() {
             if (Double.isInfinite(from)) {
@@ -216,7 +225,7 @@ public class InternalRange<B extends InternalRange.Bucket> extends InternalMulti
         }
     }
 
-    public static class Factory<B extends Bucket, R extends InternalRange<B>> {
+    public static class Factory<B extends Bucket, R extends InternalRange<B, R>> extends InternalMultiBucketAggregation.Factory<R, B> {
 
         public String type() {
             return TYPE.name();
@@ -231,12 +240,25 @@ public class InternalRange<B extends InternalRange.Bucket> extends InternalMulti
         public B createBucket(String key, double from, double to, long docCount, InternalAggregations aggregations, boolean keyed, @Nullable ValueFormatter formatter) {
             return (B) new Bucket(key, from, to, docCount, aggregations, keyed, formatter);
         }
+
+        @Override
+        public R create(List<B> ranges, R prototype) {
+            return (R) new InternalRange<>(prototype.name, ranges, prototype.formatter, prototype.keyed, prototype.reducers(),
+                    prototype.metaData);
+        }
+
+        @Override
+        public B createBucket(InternalAggregations aggregations, B prototype) {
+            return (B) new Bucket(prototype.getKey(), prototype.from, prototype.to, prototype.getDocCount(), aggregations, prototype.keyed,
+                    prototype.formatter);
+        }
     }
 
     private List<B> ranges;
     private Map<String, B> rangeMap;
-    private @Nullable ValueFormatter formatter;
-    private boolean keyed;
+    @Nullable
+    protected ValueFormatter formatter;
+    protected boolean keyed;
 
     public InternalRange() {} // for serialization
 
@@ -258,10 +280,20 @@ public class InternalRange<B extends InternalRange.Bucket> extends InternalMulti
         return ranges;
     }
 
-    protected Factory<B, ?> getFactory() {
+    public Factory<B, R> getFactory() {
         return FACTORY;
     }
 
+    @Override
+    public R create(List<B> buckets) {
+        return getFactory().create(buckets, (R) this);
+    }
+
+    @Override
+    public B createBucket(InternalAggregations aggregations, B prototype) {
+        return getFactory().createBucket(aggregations, prototype);
+    }
+
     @Override
     public InternalAggregation doReduce(ReduceContext reduceContext) {
         List<InternalAggregation> aggregations = reduceContext.aggregations();
@@ -271,7 +303,7 @@ public class InternalRange<B extends InternalRange.Bucket> extends InternalMulti
             rangeList[i] = new ArrayList<Bucket>();
         }
         for (InternalAggregation aggregation : aggregations) {
-            InternalRange<?> ranges = (InternalRange<?>) aggregation;
+            InternalRange<B, R> ranges = (InternalRange<B, R>) aggregation;
             int i = 0;
             for (Bucket range : ranges.ranges) {
                 rangeList[i++].add(range);

src/main/java/org/elasticsearch/search/aggregations/bucket/range/date/InternalDateRange.java

@@ -38,7 +38,7 @@ import java.util.Map;
 /**
  *
  */
-public class InternalDateRange extends InternalRange<InternalDateRange.Bucket> {
+public class InternalDateRange extends InternalRange<InternalDateRange.Bucket, InternalDateRange> {
 
     public final static Type TYPE = new Type("date_range", "drange");
 
@@ -113,7 +113,7 @@ public class InternalDateRange extends InternalRange<InternalDateRange.Bucket> {
         }
     }
 
-    private static class Factory extends InternalRange.Factory<InternalDateRange.Bucket, InternalDateRange> {
+    public static class Factory extends InternalRange.Factory<InternalDateRange.Bucket, InternalDateRange> {
 
         @Override
         public String type() {
@@ -126,10 +126,22 @@ public class InternalDateRange extends InternalRange<InternalDateRange.Bucket> {
             return new InternalDateRange(name, ranges, formatter, keyed, reducers, metaData);
         }
 
+        @Override
+        public InternalDateRange create(List<Bucket> ranges, InternalDateRange prototype) {
+            return new InternalDateRange(prototype.name, ranges, prototype.formatter, prototype.keyed, prototype.reducers(),
+                    prototype.metaData);
+        }
+
         @Override
         public Bucket createBucket(String key, double from, double to, long docCount, InternalAggregations aggregations, boolean keyed, ValueFormatter formatter) {
             return new Bucket(key, from, to, docCount, aggregations, keyed, formatter);
         }
+
+        @Override
+        public Bucket createBucket(InternalAggregations aggregations, Bucket prototype) {
+            return new Bucket(prototype.getKey(), ((Number) prototype.getFrom()).doubleValue(), ((Number) prototype.getTo()).doubleValue(),
+                    prototype.getDocCount(), aggregations, prototype.getKeyed(), prototype.getFormatter());
+        }
     }
 
     InternalDateRange() {} // for serialization
@@ -145,7 +157,7 @@ public class InternalDateRange extends InternalRange<InternalDateRange.Bucket> {
     }
 
     @Override
-    protected InternalRange.Factory<Bucket, ?> getFactory() {
+    public InternalRange.Factory<Bucket, InternalDateRange> getFactory() {
         return FACTORY;
     }
 }

src/main/java/org/elasticsearch/search/aggregations/bucket/range/geodistance/InternalGeoDistance.java

@@ -36,7 +36,7 @@ import java.util.Map;
 /**
  *
  */
-public class InternalGeoDistance extends InternalRange<InternalGeoDistance.Bucket> {
+public class InternalGeoDistance extends InternalRange<InternalGeoDistance.Bucket, InternalGeoDistance> {
 
     public static final Type TYPE = new Type("geo_distance", "gdist");
 
@@ -101,7 +101,7 @@ public class InternalGeoDistance extends InternalRange<InternalGeoDistance.Bucke
         }
     }
 
-    private static class Factory extends InternalRange.Factory<InternalGeoDistance.Bucket, InternalGeoDistance> {
+    public static class Factory extends InternalRange.Factory<InternalGeoDistance.Bucket, InternalGeoDistance> {
 
         @Override
         public String type() {
@@ -114,10 +114,22 @@ public class InternalGeoDistance extends InternalRange<InternalGeoDistance.Bucke
             return new InternalGeoDistance(name, ranges, formatter, keyed, reducers, metaData);
         }
 
+        @Override
+        public InternalGeoDistance create(List<Bucket> ranges, InternalGeoDistance prototype) {
+            return new InternalGeoDistance(prototype.name, ranges, prototype.formatter, prototype.keyed, prototype.reducers(),
+                    prototype.metaData);
+        }
+
         @Override
         public Bucket createBucket(String key, double from, double to, long docCount, InternalAggregations aggregations, boolean keyed, @Nullable ValueFormatter formatter) {
             return new Bucket(key, from, to, docCount, aggregations, keyed, formatter);
         }
+
+        @Override
+        public Bucket createBucket(InternalAggregations aggregations, Bucket prototype) {
+            return new Bucket(prototype.getKey(), ((Number) prototype.getFrom()).doubleValue(), ((Number) prototype.getTo()).doubleValue(),
+                    prototype.getDocCount(), aggregations, prototype.getKeyed(), prototype.getFormatter());
+        }
     }
 
     InternalGeoDistance() {} // for serialization
@@ -133,7 +145,7 @@ public class InternalGeoDistance extends InternalRange<InternalGeoDistance.Bucke
     }
 
     @Override
-    protected InternalRange.Factory<Bucket, ?> getFactory() {
+    public InternalRange.Factory<Bucket, InternalGeoDistance> getFactory() {
         return FACTORY;
     }
 }

src/main/java/org/elasticsearch/search/aggregations/bucket/range/ipv4/InternalIPv4Range.java

@@ -36,7 +36,7 @@ import java.util.Map;
 /**
  *
  */
-public class InternalIPv4Range extends InternalRange<InternalIPv4Range.Bucket> {
+public class InternalIPv4Range extends InternalRange<InternalIPv4Range.Bucket, InternalIPv4Range> {
 
     public static final long MAX_IP = 4294967296l;
 
@@ -110,7 +110,7 @@ public class InternalIPv4Range extends InternalRange<InternalIPv4Range.Bucket> {
         }
     }
 
-    private static class Factory extends InternalRange.Factory<InternalIPv4Range.Bucket, InternalIPv4Range> {
+    public static class Factory extends InternalRange.Factory<InternalIPv4Range.Bucket, InternalIPv4Range> {
 
         @Override
         public String type() {
@@ -123,10 +123,21 @@ public class InternalIPv4Range extends InternalRange<InternalIPv4Range.Bucket> {
             return new InternalIPv4Range(name, ranges, keyed, reducers, metaData);
         }
 
+        @Override
+        public InternalIPv4Range create(List<Bucket> ranges, InternalIPv4Range prototype) {
+            return new InternalIPv4Range(prototype.name, ranges, prototype.keyed, prototype.reducers(), prototype.metaData);
+        }
+
         @Override
         public Bucket createBucket(String key, double from, double to, long docCount, InternalAggregations aggregations, boolean keyed, @Nullable ValueFormatter formatter) {
             return new Bucket(key, from, to, docCount, aggregations, keyed);
         }
+
+        @Override
+        public Bucket createBucket(InternalAggregations aggregations, Bucket prototype) {
+            return new Bucket(prototype.getKey(), ((Number) prototype.getFrom()).doubleValue(), ((Number) prototype.getTo()).doubleValue(),
+                    prototype.getDocCount(), aggregations, prototype.getKeyed());
+        }
     }
 
     public InternalIPv4Range() {} // for serialization
@@ -142,7 +153,7 @@ public class InternalIPv4Range extends InternalRange<InternalIPv4Range.Bucket> {
     }
 
     @Override
-    protected InternalRange.Factory<Bucket, ?> getFactory() {
+    public InternalRange.Factory<Bucket, InternalIPv4Range> getFactory() {
         return FACTORY;
     }
 }

src/main/java/org/elasticsearch/search/aggregations/bucket/significant/InternalSignificantTerms.java

@@ -39,12 +39,13 @@ import java.util.Map;
 /**
  *
  */
-public abstract class InternalSignificantTerms extends InternalMultiBucketAggregation implements SignificantTerms, ToXContent, Streamable {
+public abstract class InternalSignificantTerms<A extends InternalSignificantTerms, B extends InternalSignificantTerms.Bucket> extends
+        InternalMultiBucketAggregation<A, B> implements SignificantTerms, ToXContent, Streamable {
 
     protected SignificanceHeuristic significanceHeuristic;
     protected int requiredSize;
     protected long minDocCount;
-    protected List<Bucket> buckets;
+    protected List<? extends Bucket> buckets;
     protected Map<String, Bucket> bucketMap;
     protected long subsetSize;
     protected long supersetSize;
@@ -124,7 +125,8 @@ public abstract class InternalSignificantTerms extends InternalMultiBucketAggreg
     }
 
     protected InternalSignificantTerms(long subsetSize, long supersetSize, String name, int requiredSize, long minDocCount,
-            SignificanceHeuristic significanceHeuristic, List<Bucket> buckets, List<Reducer> reducers, Map<String, Object> metaData) {
+            SignificanceHeuristic significanceHeuristic, List<? extends Bucket> buckets, List<Reducer> reducers,
+            Map<String, Object> metaData) {
         super(name, reducers, metaData);
         this.requiredSize = requiredSize;
         this.minDocCount = minDocCount;
@@ -166,13 +168,13 @@ public abstract class InternalSignificantTerms extends InternalMultiBucketAggreg
         // Compute the overall result set size and the corpus size using the
         // top-level Aggregations from each shard
         for (InternalAggregation aggregation : aggregations) {
-            InternalSignificantTerms terms = (InternalSignificantTerms) aggregation;
+            InternalSignificantTerms<A, B> terms = (InternalSignificantTerms<A, B>) aggregation;
             globalSubsetSize += terms.subsetSize;
             globalSupersetSize += terms.supersetSize;
         }
         Map<String, List<InternalSignificantTerms.Bucket>> buckets = new HashMap<>();
         for (InternalAggregation aggregation : aggregations) {
-            InternalSignificantTerms terms = (InternalSignificantTerms) aggregation;
+            InternalSignificantTerms<A, B> terms = (InternalSignificantTerms<A, B>) aggregation;
             for (Bucket bucket : terms.buckets) {
                 List<Bucket> existingBuckets = buckets.get(bucket.getKey());
                 if (existingBuckets == null) {
@@ -200,9 +202,10 @@ public abstract class InternalSignificantTerms extends InternalMultiBucketAggreg
         for (int i = ordered.size() - 1; i >= 0; i--) {
             list[i] = (Bucket) ordered.pop();
         }
-        return newAggregation(globalSubsetSize, globalSupersetSize, Arrays.asList(list));
+        return create(globalSubsetSize, globalSupersetSize, Arrays.asList(list), this);
     }
 
-    abstract InternalSignificantTerms newAggregation(long subsetSize, long supersetSize, List<Bucket> buckets);
+    protected abstract A create(long subsetSize, long supersetSize, List<InternalSignificantTerms.Bucket> buckets,
+            InternalSignificantTerms prototype);
 
 }

src/main/java/org/elasticsearch/search/aggregations/bucket/significant/SignificantLongTerms.java

@@ -42,7 +42,7 @@ import java.util.Map;
 /**
  *
  */
-public class SignificantLongTerms extends InternalSignificantTerms {
+public class SignificantLongTerms extends InternalSignificantTerms<SignificantLongTerms, SignificantLongTerms.Bucket> {
 
     public static final Type TYPE = new Type("significant_terms", "siglterms");
 
@@ -162,15 +162,13 @@ public class SignificantLongTerms extends InternalSignificantTerms {
             return builder;
         }
     }
-
     private ValueFormatter formatter;
 
     SignificantLongTerms() {
     } // for serialization
 
-    public SignificantLongTerms(long subsetSize, long supersetSize, String name, @Nullable ValueFormatter formatter,
- int requiredSize,
-            long minDocCount, SignificanceHeuristic significanceHeuristic, List<InternalSignificantTerms.Bucket> buckets,
+    public SignificantLongTerms(long subsetSize, long supersetSize, String name, @Nullable ValueFormatter formatter, int requiredSize,
+            long minDocCount, SignificanceHeuristic significanceHeuristic, List<? extends InternalSignificantTerms.Bucket> buckets,
             List<Reducer> reducers, Map<String, Object> metaData) {
 
         super(subsetSize, supersetSize, name, requiredSize, minDocCount, significanceHeuristic, buckets, reducers, metaData);
@@ -183,10 +181,24 @@ public class SignificantLongTerms extends InternalSignificantTerms {
     }
 
     @Override
-    InternalSignificantTerms newAggregation(long subsetSize, long supersetSize,
-            List<InternalSignificantTerms.Bucket> buckets) {
-        return new SignificantLongTerms(subsetSize, supersetSize, getName(), formatter, requiredSize, minDocCount, significanceHeuristic,
-                buckets, reducers(), getMetaData());
+    public SignificantLongTerms create(List<SignificantLongTerms.Bucket> buckets) {
+        return new SignificantLongTerms(this.subsetSize, this.supersetSize, this.name, this.formatter, this.requiredSize, this.minDocCount,
+                this.significanceHeuristic, buckets, this.reducers(), this.metaData);
+    }
+
+    @Override
+    public Bucket createBucket(InternalAggregations aggregations, SignificantLongTerms.Bucket prototype) {
+        return new Bucket(prototype.subsetDf, prototype.subsetSize, prototype.supersetDf, prototype.supersetSize, prototype.term,
+                aggregations, prototype.formatter);
+    }
+
+    @Override
+    protected SignificantLongTerms create(long subsetSize, long supersetSize,
+            List<org.elasticsearch.search.aggregations.bucket.significant.InternalSignificantTerms.Bucket> buckets,
+            InternalSignificantTerms prototype) {
+        return new SignificantLongTerms(subsetSize, supersetSize, prototype.getName(), ((SignificantLongTerms) prototype).formatter,
+                prototype.requiredSize, prototype.minDocCount, prototype.significanceHeuristic, buckets, prototype.reducers(),
+                prototype.getMetaData());
     }
 
     @Override

src/main/java/org/elasticsearch/search/aggregations/bucket/significant/SignificantStringTerms.java

@@ -41,7 +41,7 @@ import java.util.Map;
 /**
  *
  */
-public class SignificantStringTerms extends InternalSignificantTerms {
+public class SignificantStringTerms extends InternalSignificantTerms<SignificantStringTerms, SignificantStringTerms.Bucket> {
 
     public static final InternalAggregation.Type TYPE = new Type("significant_terms", "sigsterms");
 
@@ -160,9 +160,8 @@ public class SignificantStringTerms extends InternalSignificantTerms {
 
     SignificantStringTerms() {} // for serialization
 
-    public SignificantStringTerms(long subsetSize, long supersetSize, String name, int requiredSize,
- long minDocCount,
-            SignificanceHeuristic significanceHeuristic, List<InternalSignificantTerms.Bucket> buckets, List<Reducer> reducers,
+    public SignificantStringTerms(long subsetSize, long supersetSize, String name, int requiredSize, long minDocCount,
+            SignificanceHeuristic significanceHeuristic, List<? extends InternalSignificantTerms.Bucket> buckets, List<Reducer> reducers,
             Map<String, Object> metaData) {
         super(subsetSize, supersetSize, name, requiredSize, minDocCount, significanceHeuristic, buckets, reducers, metaData);
     }
@@ -173,10 +172,22 @@ public class SignificantStringTerms extends InternalSignificantTerms {
     }
 
     @Override
-    InternalSignificantTerms newAggregation(long subsetSize, long supersetSize,
-            List<InternalSignificantTerms.Bucket> buckets) {
-        return new SignificantStringTerms(subsetSize, supersetSize, getName(), requiredSize, minDocCount, significanceHeuristic, buckets,
-                reducers(), getMetaData());
+    public SignificantStringTerms create(List<SignificantStringTerms.Bucket> buckets) {
+        return new SignificantStringTerms(this.subsetSize, this.supersetSize, this.name, this.requiredSize, this.minDocCount,
+                this.significanceHeuristic, buckets, this.reducers(), this.metaData);
+    }
+
+    @Override
+    public Bucket createBucket(InternalAggregations aggregations, SignificantStringTerms.Bucket prototype) {
+        return new Bucket(prototype.termBytes, prototype.subsetDf, prototype.subsetSize, prototype.supersetDf, prototype.supersetSize,
+                aggregations);
+    }
+
+    @Override
+    protected SignificantStringTerms create(long subsetSize, long supersetSize, List<InternalSignificantTerms.Bucket> buckets,
+            InternalSignificantTerms prototype) {
+        return new SignificantStringTerms(subsetSize, supersetSize, prototype.getName(), prototype.requiredSize, prototype.minDocCount,
+                prototype.significanceHeuristic, buckets, prototype.reducers(), prototype.getMetaData());
     }
 
     @Override

src/main/java/org/elasticsearch/search/aggregations/bucket/significant/UnmappedSignificantTerms.java

@@ -23,6 +23,7 @@ import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.search.aggregations.AggregationStreams;
 import org.elasticsearch.search.aggregations.InternalAggregation;
+import org.elasticsearch.search.aggregations.InternalAggregations;
 import org.elasticsearch.search.aggregations.bucket.significant.heuristics.JLHScore;
 import org.elasticsearch.search.aggregations.reducers.Reducer;
 
@@ -34,7 +35,7 @@ import java.util.Map;
 /**
  *
  */
-public class UnmappedSignificantTerms extends InternalSignificantTerms {
+public class UnmappedSignificantTerms extends InternalSignificantTerms<UnmappedSignificantTerms, InternalSignificantTerms.Bucket> {
 
     public static final Type TYPE = new Type("significant_terms", "umsigterms");
 
@@ -67,6 +68,21 @@ public class UnmappedSignificantTerms extends InternalSignificantTerms {
         return TYPE;
     }
 
+    @Override
+    public UnmappedSignificantTerms create(List<InternalSignificantTerms.Bucket> buckets) {
+        return new UnmappedSignificantTerms(this.name, this.requiredSize, this.minDocCount, this.reducers(), this.metaData);
+    }
+
+    @Override
+    public InternalSignificantTerms.Bucket createBucket(InternalAggregations aggregations, InternalSignificantTerms.Bucket prototype) {
+        throw new UnsupportedOperationException("not supported for UnmappedSignificantTerms");
+    }
+
+    @Override
+    protected UnmappedSignificantTerms create(long subsetSize, long supersetSize, List<Bucket> buckets, InternalSignificantTerms prototype) {
+        throw new UnsupportedOperationException("not supported for UnmappedSignificantTerms");
+    }
+
     @Override
     public InternalAggregation doReduce(ReduceContext reduceContext) {
         for (InternalAggregation aggregation : reduceContext.aggregations()) {
@@ -77,11 +93,6 @@ public class UnmappedSignificantTerms extends InternalSignificantTerms {
         return this;
     }
 
-    @Override
-    InternalSignificantTerms newAggregation(long subsetSize, long supersetSize, List<Bucket> buckets) {
-        throw new UnsupportedOperationException("How did you get there?");
-    }
-
     @Override
     protected void doReadFrom(StreamInput in) throws IOException {
         this.requiredSize = readSize(in);

src/main/java/org/elasticsearch/search/aggregations/bucket/terms/DoubleTerms.java

@@ -40,7 +40,7 @@ import java.util.Map;
 /**
  *
  */
-public class DoubleTerms extends InternalTerms {
+public class DoubleTerms extends InternalTerms<DoubleTerms, DoubleTerms.Bucket> {
 
     public static final Type TYPE = new Type("terms", "dterms");
 
@@ -85,7 +85,8 @@ public class DoubleTerms extends InternalTerms {
             super(formatter, showDocCountError);
         }
 
-        public Bucket(double term, long docCount, InternalAggregations aggregations, boolean showDocCountError, long docCountError, @Nullable ValueFormatter formatter) {
+        public Bucket(double term, long docCount, InternalAggregations aggregations, boolean showDocCountError, long docCountError,
+                @Nullable ValueFormatter formatter) {
             super(docCount, aggregations, showDocCountError, docCountError, formatter);
             this.term = term;
         }
@@ -153,13 +154,15 @@ public class DoubleTerms extends InternalTerms {
         }
     }
 
-    private @Nullable ValueFormatter formatter;
+    private @Nullable
+    ValueFormatter formatter;
 
-    DoubleTerms() {} // for serialization
+    DoubleTerms() {
+    } // for serialization
 
     public DoubleTerms(String name, Terms.Order order, @Nullable ValueFormatter formatter, int requiredSize, int shardSize,
-            long minDocCount, List<InternalTerms.Bucket> buckets, boolean showTermDocCountError, long docCountError, long otherDocCount,
-            List<Reducer> reducers, Map<String, Object> metaData) {
+            long minDocCount, List<? extends InternalTerms.Bucket> buckets, boolean showTermDocCountError, long docCountError,
+            long otherDocCount, List<Reducer> reducers, Map<String, Object> metaData) {
         super(name, order, requiredSize, shardSize, minDocCount, buckets, showTermDocCountError, docCountError, otherDocCount, reducers,
                 metaData);
         this.formatter = formatter;
@@ -171,10 +174,23 @@ public class DoubleTerms extends InternalTerms {
     }
 
     @Override
-    protected InternalTerms newAggregation(String name, List<InternalTerms.Bucket> buckets, boolean showTermDocCountError,
-            long docCountError, long otherDocCount, List<Reducer> reducers, Map<String, Object> metaData) {
-        return new DoubleTerms(name, order, formatter, requiredSize, shardSize, minDocCount, buckets, showTermDocCountError, docCountError,
-                otherDocCount, reducers, metaData);
+    public DoubleTerms create(List<Bucket> buckets) {
+        return new DoubleTerms(this.name, this.order, this.formatter, this.requiredSize, this.shardSize, this.minDocCount, buckets,
+                this.showTermDocCountError, this.docCountError, this.otherDocCount, this.reducers(), this.metaData);
+    }
+
+    @Override
+    public Bucket createBucket(InternalAggregations aggregations, Bucket prototype) {
+        return new Bucket(prototype.term, prototype.docCount, aggregations, prototype.showDocCountError, prototype.docCountError,
+                prototype.formatter);
+    }
+
+    @Override
+    protected DoubleTerms create(String name, List<org.elasticsearch.search.aggregations.bucket.terms.InternalTerms.Bucket> buckets,
+            long docCountError, long otherDocCount, InternalTerms prototype) {
+        return new DoubleTerms(name, prototype.order, ((DoubleTerms) prototype).formatter, prototype.requiredSize, prototype.shardSize,
+                prototype.minDocCount, buckets, prototype.showTermDocCountError, docCountError, otherDocCount, prototype.reducers(),
+                prototype.getMetaData());
     }
 
     @Override

src/main/java/org/elasticsearch/search/aggregations/bucket/terms/InternalTerms.java

@@ -43,7 +43,8 @@ import java.util.Map;
 /**
  *
  */
-public abstract class InternalTerms extends InternalMultiBucketAggregation implements Terms, ToXContent, Streamable {
+public abstract class InternalTerms<A extends InternalTerms, B extends InternalTerms.Bucket> extends InternalMultiBucketAggregation<A, B>
+        implements Terms, ToXContent, Streamable {
 
     protected static final String DOC_COUNT_ERROR_UPPER_BOUND_FIELD_NAME = "doc_count_error_upper_bound";
     protected static final String SUM_OF_OTHER_DOC_COUNTS = "sum_other_doc_count";
@@ -115,7 +116,7 @@ public abstract class InternalTerms extends InternalMultiBucketAggregation imple
     protected int requiredSize;
     protected int shardSize;
     protected long minDocCount;
-    protected List<Bucket> buckets;
+    protected List<? extends Bucket> buckets;
     protected Map<String, Bucket> bucketMap;
     protected long docCountError;
     protected boolean showTermDocCountError;
@@ -123,8 +124,9 @@ public abstract class InternalTerms extends InternalMultiBucketAggregation imple
 
     protected InternalTerms() {} // for serialization
 
-    protected InternalTerms(String name, Terms.Order order, int requiredSize, int shardSize, long minDocCount, List<Bucket> buckets,
-            boolean showTermDocCountError, long docCountError, long otherDocCount, List<Reducer> reducers, Map<String, Object> metaData) {
+    protected InternalTerms(String name, Terms.Order order, int requiredSize, int shardSize, long minDocCount,
+            List<? extends Bucket> buckets, boolean showTermDocCountError, long docCountError, long otherDocCount, List<Reducer> reducers,
+            Map<String, Object> metaData) {
         super(name, reducers, metaData);
         this.order = order;
         this.requiredSize = requiredSize;
@@ -171,7 +173,7 @@ public abstract class InternalTerms extends InternalMultiBucketAggregation imple
         long sumDocCountError = 0;
         long otherDocCount = 0;
         for (InternalAggregation aggregation : aggregations) {
-            InternalTerms terms = (InternalTerms) aggregation;
+            InternalTerms<A, B> terms = (InternalTerms<A, B>) aggregation;
             otherDocCount += terms.getSumOfOtherDocCounts();
             final long thisAggDocCountError;
             if (terms.buckets.size() < this.shardSize || this.order == InternalOrder.TERM_ASC || this.order == InternalOrder.TERM_DESC) {
@@ -224,10 +226,10 @@ public abstract class InternalTerms extends InternalMultiBucketAggregation imple
         } else {
             docCountError = aggregations.size() == 1 ? 0 : sumDocCountError;
         }
-        return newAggregation(name, Arrays.asList(list), showTermDocCountError, docCountError, otherDocCount, reducers(), getMetaData());
+        return create(name, Arrays.asList(list), docCountError, otherDocCount, this);
     }
 
-    protected abstract InternalTerms newAggregation(String name, List<Bucket> buckets, boolean showTermDocCountError, long docCountError,
-            long otherDocCount, List<Reducer> reducers, Map<String, Object> metaData);
+    protected abstract A create(String name, List<InternalTerms.Bucket> buckets, long docCountError, long otherDocCount,
+            InternalTerms prototype);
 
 }

src/main/java/org/elasticsearch/search/aggregations/bucket/terms/LongTerms.java

@@ -39,7 +39,7 @@ import java.util.Map;
 /**
  *
  */
-public class LongTerms extends InternalTerms {
+public class LongTerms extends InternalTerms<LongTerms, LongTerms.Bucket> {
 
     public static final Type TYPE = new Type("terms", "lterms");
 
@@ -157,7 +157,7 @@ public class LongTerms extends InternalTerms {
     LongTerms() {} // for serialization
 
     public LongTerms(String name, Terms.Order order, @Nullable ValueFormatter formatter, int requiredSize, int shardSize, long minDocCount,
-            List<InternalTerms.Bucket> buckets, boolean showTermDocCountError, long docCountError, long otherDocCount,
+            List<? extends InternalTerms.Bucket> buckets, boolean showTermDocCountError, long docCountError, long otherDocCount,
             List<Reducer> reducers, Map<String, Object> metaData) {
         super(name, order, requiredSize, shardSize, minDocCount, buckets, showTermDocCountError, docCountError, otherDocCount, reducers,
                 metaData);
@@ -170,10 +170,23 @@ public class LongTerms extends InternalTerms {
     }
 
     @Override
-    protected InternalTerms newAggregation(String name, List<InternalTerms.Bucket> buckets, boolean showTermDocCountError,
-            long docCountError, long otherDocCount, List<Reducer> reducers, Map<String, Object> metaData) {
-        return new LongTerms(name, order, formatter, requiredSize, shardSize, minDocCount, buckets, showTermDocCountError, docCountError,
-                otherDocCount, reducers, metaData);
+    public LongTerms create(List<Bucket> buckets) {
+        return new LongTerms(this.name, this.order, this.formatter, this.requiredSize, this.shardSize, this.minDocCount, buckets,
+                this.showTermDocCountError, this.docCountError, this.otherDocCount, this.reducers(), this.metaData);
+    }
+
+    @Override
+    public Bucket createBucket(InternalAggregations aggregations, Bucket prototype) {
+        return new Bucket(prototype.term, prototype.docCount, aggregations, prototype.showDocCountError, prototype.docCountError,
+                prototype.formatter);
+    }
+
+    @Override
+    protected LongTerms create(String name, List<org.elasticsearch.search.aggregations.bucket.terms.InternalTerms.Bucket> buckets,
+            long docCountError, long otherDocCount, InternalTerms prototype) {
+        return new LongTerms(name, prototype.order, ((LongTerms) prototype).formatter, prototype.requiredSize, prototype.shardSize,
+                prototype.minDocCount, buckets, prototype.showTermDocCountError, docCountError, otherDocCount, prototype.reducers(),
+                prototype.getMetaData());
     }
 
     @Override

src/main/java/org/elasticsearch/search/aggregations/bucket/terms/StringTerms.java

@@ -38,7 +38,7 @@ import java.util.Map;
 /**
  *
  */
-public class StringTerms extends InternalTerms {
+public class StringTerms extends InternalTerms<StringTerms, StringTerms.Bucket> {
 
     public static final InternalAggregation.Type TYPE = new Type("terms", "sterms");
 
@@ -74,7 +74,6 @@ public class StringTerms extends InternalTerms {
         BucketStreams.registerStream(BUCKET_STREAM, TYPE.stream());
     }
 
-
     public static class Bucket extends InternalTerms.Bucket {
 
         BytesRef termBytes;
@@ -149,10 +148,11 @@ public class StringTerms extends InternalTerms {
         }
     }
 
-    StringTerms() {} // for serialization
+    StringTerms() {
+    } // for serialization
 
     public StringTerms(String name, Terms.Order order, int requiredSize, int shardSize, long minDocCount,
-            List<InternalTerms.Bucket> buckets, boolean showTermDocCountError, long docCountError, long otherDocCount,
+            List<? extends InternalTerms.Bucket> buckets, boolean showTermDocCountError, long docCountError, long otherDocCount,
             List<Reducer> reducers, Map<String, Object> metaData) {
         super(name, order, requiredSize, shardSize, minDocCount, buckets, showTermDocCountError, docCountError, otherDocCount, reducers,
                 metaData);
@@ -164,10 +164,21 @@ public class StringTerms extends InternalTerms {
     }
 
     @Override
-    protected InternalTerms newAggregation(String name, List<InternalTerms.Bucket> buckets, boolean showTermDocCountError,
-            long docCountError, long otherDocCount, List<Reducer> reducers, Map<String, Object> metaData) {
-        return new StringTerms(name, order, requiredSize, shardSize, minDocCount, buckets, showTermDocCountError, docCountError,
-                otherDocCount, reducers, metaData);
+    public StringTerms create(List<Bucket> buckets) {
+        return new StringTerms(this.name, this.order, this.requiredSize, this.shardSize, this.minDocCount, buckets,
+                this.showTermDocCountError, this.docCountError, this.otherDocCount, this.reducers(), this.metaData);
+    }
+
+    @Override
+    public Bucket createBucket(InternalAggregations aggregations, Bucket prototype) {
+        return new Bucket(prototype.termBytes, prototype.docCount, aggregations, prototype.showDocCountError, prototype.docCountError);
+    }
+
+    @Override
+    protected StringTerms create(String name, List<org.elasticsearch.search.aggregations.bucket.terms.InternalTerms.Bucket> buckets,
+            long docCountError, long otherDocCount, InternalTerms prototype) {
+        return new StringTerms(name, prototype.order, prototype.requiredSize, prototype.shardSize, prototype.minDocCount, buckets,
+                prototype.showTermDocCountError, docCountError, otherDocCount, prototype.reducers(), prototype.getMetaData());
     }
 
     @Override

src/main/java/org/elasticsearch/search/aggregations/bucket/terms/UnmappedTerms.java

@@ -23,6 +23,7 @@ import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.search.aggregations.AggregationStreams;
 import org.elasticsearch.search.aggregations.InternalAggregation;
+import org.elasticsearch.search.aggregations.InternalAggregations;
 import org.elasticsearch.search.aggregations.reducers.Reducer;
 
 import java.io.IOException;
@@ -33,7 +34,7 @@ import java.util.Map;
 /**
  *
  */
-public class UnmappedTerms extends InternalTerms {
+public class UnmappedTerms extends InternalTerms<UnmappedTerms, InternalTerms.Bucket> {
 
     public static final Type TYPE = new Type("terms", "umterms");
 
@@ -65,6 +66,21 @@ public class UnmappedTerms extends InternalTerms {
         return TYPE;
     }
 
+    @Override
+    public UnmappedTerms create(List<InternalTerms.Bucket> buckets) {
+        return new UnmappedTerms(this.name, this.order, this.requiredSize, this.shardSize, this.minDocCount, this.reducers(), this.metaData);
+    }
+
+    @Override
+    public InternalTerms.Bucket createBucket(InternalAggregations aggregations, InternalTerms.Bucket prototype) {
+        throw new UnsupportedOperationException("not supported for UnmappedTerms");
+    }
+
+    @Override
+    protected UnmappedTerms create(String name, List<Bucket> buckets, long docCountError, long otherDocCount, InternalTerms prototype) {
+        throw new UnsupportedOperationException("not supported for UnmappedTerms");
+    }
+
     @Override
     protected void doReadFrom(StreamInput in) throws IOException {
         this.docCountError = 0;
@@ -92,12 +108,6 @@ public class UnmappedTerms extends InternalTerms {
         return this;
     }
 
-    @Override
-    protected InternalTerms newAggregation(String name, List<Bucket> buckets, boolean showTermDocCountError, long docCountError,
-            long otherDocCount, List<Reducer> reducers, Map<String, Object> metaData) {
-        throw new UnsupportedOperationException("How did you get there?");
-    }
-
     @Override
     public XContentBuilder doXContentBody(XContentBuilder builder, Params params) throws IOException {
         builder.field(InternalTerms.DOC_COUNT_ERROR_UPPER_BOUND_FIELD_NAME, docCountError);

src/main/java/org/elasticsearch/search/aggregations/reducers/derivative/DerivativeReducer.java

@@ -95,7 +95,7 @@ public class DerivativeReducer extends Reducer {
 
     @Override
     public InternalAggregation reduce(InternalAggregation aggregation, ReduceContext reduceContext) {
-        InternalHistogram<? extends InternalHistogram.Bucket> histo = (InternalHistogram<? extends InternalHistogram.Bucket>) aggregation;
+        InternalHistogram histo = (InternalHistogram) aggregation;
         List<? extends InternalHistogram.Bucket> buckets = histo.getBuckets();
         InternalHistogram.Factory<? extends InternalHistogram.Bucket> factory = histo.getFactory();
 
@@ -116,7 +116,7 @@ public class DerivativeReducer extends Reducer {
             }
             lastBucketValue = thisBucketValue;
         }
-        return factory.create(histo.getName(), newBuckets, histo);
+        return factory.create(newBuckets, histo);
     }
 
     @Override