derivative reducer now works with both date_histogram and histogram

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

@@ -19,6 +19,7 @@
 package org.elasticsearch.search.aggregations.bucket.histogram;
 
 import org.elasticsearch.common.Nullable;
+import org.elasticsearch.search.aggregations.AggregationExecutionException;
 import org.elasticsearch.search.aggregations.InternalAggregation.Type;
 import org.elasticsearch.search.aggregations.InternalAggregations;
 import org.elasticsearch.search.aggregations.bucket.histogram.InternalHistogram.EmptyBucketInfo;
@@ -83,8 +84,15 @@ public class InternalDateHistogram {
         }
 
         @Override
-        public InternalDateHistogram.Bucket createBucket(long key, long docCount, InternalAggregations aggregations, boolean keyed, @Nullable ValueFormatter formatter) {
+        public InternalDateHistogram.Bucket createBucket(Object key, long docCount, InternalAggregations aggregations, boolean keyed,
-            return new Bucket(key, docCount, aggregations, keyed, formatter, this);
+                @Nullable ValueFormatter formatter) {
+            if (key instanceof Number) {
+                return new Bucket(((Number) key).longValue(), docCount, aggregations, keyed, formatter, this);
+            } else if (key instanceof DateTime) {
+                return new Bucket(((DateTime) key).getMillis(), docCount, aggregations, keyed, formatter, this);
+            } else {
+                throw new AggregationExecutionException("Expected key of type Number or DateTime but got [" + key + "]");
+            }
         }
 
         @Override

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

@@ -30,6 +30,7 @@ import org.elasticsearch.common.rounding.Rounding;
 import org.elasticsearch.common.text.StringText;
 import org.elasticsearch.common.text.Text;
 import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.search.aggregations.AggregationExecutionException;
 import org.elasticsearch.search.aggregations.AggregationStreams;
 import org.elasticsearch.search.aggregations.Aggregations;
 import org.elasticsearch.search.aggregations.InternalAggregation;
@@ -247,8 +248,13 @@ public class InternalHistogram<B extends InternalHistogram.Bucket> extends Inter
             return new InternalHistogram<>(name, buckets, order, minDocCount, emptyBucketInfo, formatter, keyed, this, reducers, metaData);
         }
 
-        public B createBucket(long key, long docCount, InternalAggregations aggregations, boolean keyed, @Nullable ValueFormatter formatter) {
+        public B createBucket(Object key, long docCount, InternalAggregations aggregations, boolean keyed,
-            return (B) new Bucket(key, docCount, keyed, formatter, this, aggregations);
+                @Nullable ValueFormatter formatter) {
+            if (key instanceof Number) {
+                return (B) new Bucket(((Number) key).longValue(), docCount, keyed, formatter, this, aggregations);
+            } else {
+                throw new AggregationExecutionException("Expected key of type Number but got [" + key + "]");
+            }
         }
 
         protected B createEmptyBucket(boolean keyed, @Nullable ValueFormatter formatter) {

src/main/java/org/elasticsearch/search/aggregations/reducers/ReducerFactory.java

@@ -49,7 +49,7 @@ public abstract class ReducerFactory {
     /**
      * Validates the state of this factory (makes sure the factory is properly configured)
      */
-    public final void validate() {
+    public final void validate() { // NOCOMMIT hook in validation 
         doValidate();
     }
 

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

@@ -37,7 +37,6 @@ import org.elasticsearch.search.aggregations.reducers.ReducerFactory;
 import org.elasticsearch.search.aggregations.reducers.ReducerStreams;
 import org.elasticsearch.search.aggregations.support.AggregationContext;
 import org.elasticsearch.search.aggregations.support.AggregationPath;
-import org.joda.time.DateTime;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -89,6 +88,7 @@ public class DerivativeReducer extends Reducer {
         InternalHistogram.Factory<? extends InternalHistogram.Bucket> factory = histo.getFactory();
         List newBuckets = new ArrayList<>();
         Double lastBucketValue = null;
+        // NOCOMMIT this needs to be improved so that the aggs are cloned correctly to ensure aggs are fully immutable.
         for (InternalHistogram.Bucket bucket : buckets) {
             double thisBucketValue = (double) bucket.getProperty(histo.getName(), AggregationPath.parse(bucketsPath)
                     .getPathElementsAsStringList());
@@ -97,8 +97,9 @@ public class DerivativeReducer extends Reducer {
 
                 List<InternalAggregation> aggs = new ArrayList<>(Lists.transform(bucket.getAggregations().asList(), FUNCTION));
                 aggs.add(new InternalSimpleValue(name(), diff, null, new ArrayList<Reducer>(), metaData())); // NOCOMMIT implement formatter for derivative reducer
-                InternalHistogram.Bucket newBucket = factory.createBucket(((DateTime) bucket.getKey()).getMillis(), bucket.getDocCount(),
+                InternalHistogram.Bucket newBucket = factory.createBucket(bucket.getKey(), bucket.getDocCount(),
-                        new InternalAggregations(aggs), bucket.getKeyed(), bucket.getFormatter()); // NOCOMMIT fix key resolution to deal with numbers and dates
+ new InternalAggregations(
+                        aggs), bucket.getKeyed(), bucket.getFormatter());
                 newBuckets.add(newBucket);
             } else {
                 newBuckets.add(bucket);