Fix NPE in Derivative Pipeline when current bucket is null

Sequence of events that lead to the NPE:

- avg metric returns NaN for buckets
- Movavg skips NaN or null buckets, and simply re-uses the existing bucket (e.g. doesn't add
a 'movavg' field)
- Derivative references Movavg, the bucket resolution returns null because Movavg wasn't added
to the bucket, NPE when trying to subtract null values

Author: polyfractal

core/src/main/java/org/elasticsearch/search/aggregations/pipeline/derivative/DerivativePipelineAggregator.java

@@ -92,7 +92,7 @@ public InternalAggregation reduce(InternalAggregation aggregation, ReduceContext
         for (InternalHistogram.Bucket bucket : buckets) {
             Long thisBucketKey = resolveBucketKeyAsLong(bucket);
             Double thisBucketValue = resolveBucketValue(histo, bucket, bucketsPaths()[0], gapPolicy);
-            if (lastBucketValue != null) {
+            if (lastBucketValue != null && thisBucketValue != null) {
                 double gradient = thisBucketValue - lastBucketValue;
                 double xDiff = -1;
                 if (xAxisUnits != null) {

core/src/test/java/org/elasticsearch/search/aggregations/pipeline/DerivativeIT.java

@@ -33,6 +33,7 @@
 import org.elasticsearch.search.aggregations.metrics.sum.Sum;
 import org.elasticsearch.search.aggregations.pipeline.BucketHelpers.GapPolicy;
 import org.elasticsearch.search.aggregations.pipeline.derivative.Derivative;
+import org.elasticsearch.search.aggregations.pipeline.movavg.models.SimpleModel;
 import org.elasticsearch.search.aggregations.support.AggregationPath;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.hamcrest.Matchers;
@@ -43,11 +44,9 @@
 
 import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
 import static org.elasticsearch.index.query.QueryBuilders.matchAllQuery;
-import static org.elasticsearch.search.aggregations.AggregationBuilders.filters;
-import static org.elasticsearch.search.aggregations.AggregationBuilders.histogram;
-import static org.elasticsearch.search.aggregations.AggregationBuilders.stats;
-import static org.elasticsearch.search.aggregations.AggregationBuilders.sum;
+import static org.elasticsearch.search.aggregations.AggregationBuilders.*;
 import static org.elasticsearch.search.aggregations.pipeline.PipelineAggregatorBuilders.derivative;
+import static org.elasticsearch.search.aggregations.pipeline.PipelineAggregatorBuilders.movingAvg;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertSearchResponse;
 import static org.hamcrest.Matchers.closeTo;
@@ -614,6 +613,37 @@ public void testSingleValueAggDerivative_invalidPath() throws Exception {
         }
     }
 
+    public void testAvgMovavgDerivNPE() throws Exception {
+        createIndex("movavg_npe");
+        ensureYellow("movavg_npe");
+
+        for (int i = 0; i < 10; i++) {
+            Integer value = i;
+            if (i == 1 || i == 3) {
+                value = null;
+            }
+
+            XContentBuilder doc = jsonBuilder()
+                    .startObject()
+                    .field("tick", i)
+                    .field("value", value)
+                    .endObject();
+            client().prepareIndex("movavg_npe", "type").setSource(doc).get();
+        }
+
+        refresh();
+
+        SearchResponse response = client()
+                .prepareSearch("movavg_npe")
+                .addAggregation(
+                        histogram("histo").field("tick").interval(1)
+                                .subAggregation(avg("avg").field("value"))
+                                .subAggregation(movingAvg("movavg", "avg").modelBuilder(new SimpleModel.SimpleModelBuilder()).window(3))
+                                .subAggregation(derivative("deriv", "movavg"))).execute().actionGet();
+
+        assertSearchResponse(response);
+    }
+
     private void checkBucketKeyAndDocCount(final String msg, final Histogram.Bucket bucket, final long expectedKey,
             final long expectedDocCount) {
         assertThat(msg, bucket, notNullValue());