Fix NPE when CumulativeSum agg encounters null/empty bucket (#29641)

Fix NPE when CumulativeSum agg encounters null/empty bucket

If the cusum agg encounters a null value, it's because the value is
missing (like the first value from a derivative agg), the path is
not valid, or the bucket in the path was empty.

Previously cusum would just explode on the null, but this changes it
so we only increment the sum if the value is non-null and finite.
This is safe because even if the cusum encounters all null or empty
buckets, the cumulative sum is still zero (like how the sum agg returns
zero even if all the docs were missing values)

I went ahead and tweaked AggregatorTestCase to allow testing pipelines,
so that I could delete the IT test and reimplement it as AggTests.

Closes #27544

Author: polyfractal

docs/CHANGELOG.asciidoc

@@ -70,6 +70,8 @@ Fixed prerelease version of elasticsearch in the `deb` package to sort before GA
 Fail snapshot operations early when creating or deleting a snapshot on a repository that has been
 written to by an older Elasticsearch after writing to it with a newer Elasticsearch version. ({pull}30140[#30140])
 
+Fix NPE when CumulativeSum agg encounters null value/empty bucket ({pull}29641[#29641])
+
 //[float]
 //=== Regressions
 
@@ -100,6 +102,7 @@ multi-argument versions. ({pull}29623[#29623])
 
 Do not ignore request analysis/similarity settings on index resize operations when the source index already contains such settings ({pull}30216[#30216])
 
+Fix NPE when CumulativeSum agg encounters null value/empty bucket ({pull}29641[#29641])
 
 //[float]
 //=== Regressions

server/src/main/java/org/elasticsearch/search/aggregations/pipeline/cumulativesum/CumulativeSumPipelineAggregator.java

@@ -79,11 +79,17 @@ public InternalAggregation reduce(InternalAggregation aggregation, ReduceContext
         double sum = 0;
         for (InternalMultiBucketAggregation.InternalBucket bucket : buckets) {
             Double thisBucketValue = resolveBucketValue(histo, bucket, bucketsPaths()[0], GapPolicy.INSERT_ZEROS);
-            sum += thisBucketValue;
-            List<InternalAggregation> aggs = StreamSupport.stream(bucket.getAggregations().spliterator(), false).map((p) -> {
-                return (InternalAggregation) p;
-            }).collect(Collectors.toList());
-            aggs.add(new InternalSimpleValue(name(), sum, formatter, new ArrayList<PipelineAggregator>(), metaData()));
+
+            // Only increment the sum if it's a finite value, otherwise "increment by zero" is correct
+            if (thisBucketValue != null && thisBucketValue.isInfinite() == false && thisBucketValue.isNaN() == false) {
+                sum += thisBucketValue;
+            }
+
+            List<InternalAggregation> aggs = StreamSupport
+                .stream(bucket.getAggregations().spliterator(), false)
+                .map((p) -> (InternalAggregation) p)
+                .collect(Collectors.toList());
+            aggs.add(new InternalSimpleValue(name(), sum, formatter, new ArrayList<>(), metaData()));
             Bucket newBucket = factory.createBucket(factory.getKey(bucket), bucket.getDocCount(), new InternalAggregations(aggs));
             newBuckets.add(newBucket);
         }

server/src/test/java/org/elasticsearch/search/aggregations/pipeline/CumulativeSumAggregatorTests.java

@@ -0,0 +1,316 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.search.aggregations.pipeline;
+
+import org.apache.lucene.document.Document;
+import org.apache.lucene.document.NumericDocValuesField;
+import org.apache.lucene.document.SortedNumericDocValuesField;
+import org.apache.lucene.index.DirectoryReader;
+import org.apache.lucene.index.IndexReader;
+import org.apache.lucene.index.RandomIndexWriter;
+import org.apache.lucene.search.IndexSearcher;
+import org.apache.lucene.search.MatchAllDocsQuery;
+import org.apache.lucene.search.MatchNoDocsQuery;
+import org.apache.lucene.search.Query;
+import org.apache.lucene.store.Directory;
+import org.elasticsearch.common.CheckedConsumer;
+import org.elasticsearch.index.mapper.DateFieldMapper;
+import org.elasticsearch.index.mapper.MappedFieldType;
+import org.elasticsearch.index.mapper.NumberFieldMapper;
+import org.elasticsearch.search.aggregations.AggregationBuilder;
+import org.elasticsearch.search.aggregations.AggregatorTestCase;
+import org.elasticsearch.search.aggregations.InternalAggregation;
+import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramAggregationBuilder;
+import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval;
+import org.elasticsearch.search.aggregations.bucket.histogram.Histogram;
+import org.elasticsearch.search.aggregations.bucket.histogram.HistogramAggregationBuilder;
+import org.elasticsearch.search.aggregations.metrics.avg.AvgAggregationBuilder;
+import org.elasticsearch.search.aggregations.metrics.avg.InternalAvg;
+import org.elasticsearch.search.aggregations.metrics.sum.Sum;
+import org.elasticsearch.search.aggregations.metrics.sum.SumAggregationBuilder;
+import org.elasticsearch.search.aggregations.pipeline.cumulativesum.CumulativeSumPipelineAggregationBuilder;
+import org.elasticsearch.search.aggregations.pipeline.derivative.DerivativePipelineAggregationBuilder;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.function.Consumer;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.core.IsNull.notNullValue;
+
+public class CumulativeSumAggregatorTests extends AggregatorTestCase {
+
+    private static final String HISTO_FIELD = "histo";
+    private static final String VALUE_FIELD = "value_field";
+
+    private static final List<String> datasetTimes = Arrays.asList(
+        "2017-01-01T01:07:45",
+        "2017-01-02T03:43:34",
+        "2017-01-03T04:11:00",
+        "2017-01-04T05:11:31",
+        "2017-01-05T08:24:05",
+        "2017-01-06T13:09:32",
+        "2017-01-07T13:47:43",
+        "2017-01-08T16:14:34",
+        "2017-01-09T17:09:50",
+        "2017-01-10T22:55:46");
+
+    private static final List<Integer> datasetValues = Arrays.asList(1,2,3,4,5,6,7,8,9,10);
+
+    public void testSimple() throws IOException {
+        Query query = new MatchAllDocsQuery();
+
+        DateHistogramAggregationBuilder aggBuilder = new DateHistogramAggregationBuilder("histo");
+        aggBuilder.dateHistogramInterval(DateHistogramInterval.DAY).field(HISTO_FIELD);
+        aggBuilder.subAggregation(new AvgAggregationBuilder("the_avg").field(VALUE_FIELD));
+        aggBuilder.subAggregation(new CumulativeSumPipelineAggregationBuilder("cusum", "the_avg"));
+
+        executeTestCase(query, aggBuilder, histogram -> {
+            assertEquals(10, ((Histogram)histogram).getBuckets().size());
+            List<? extends Histogram.Bucket> buckets = ((Histogram)histogram).getBuckets();
+            double sum = 0.0;
+            for (Histogram.Bucket bucket : buckets) {
+                sum += ((InternalAvg) (bucket.getAggregations().get("the_avg"))).value();
+                assertThat(((InternalSimpleValue) (bucket.getAggregations().get("cusum"))).value(), equalTo(sum));
+            }
+        });
+    }
+
+    /**
+     * First value from a derivative is null, so this makes sure the cusum can handle that
+     */
+    public void testDerivative() throws IOException {
+        Query query = new MatchAllDocsQuery();
+
+        DateHistogramAggregationBuilder aggBuilder = new DateHistogramAggregationBuilder("histo");
+        aggBuilder.dateHistogramInterval(DateHistogramInterval.DAY).field(HISTO_FIELD);
+        aggBuilder.subAggregation(new AvgAggregationBuilder("the_avg").field(VALUE_FIELD));
+        aggBuilder.subAggregation(new DerivativePipelineAggregationBuilder("the_deriv", "the_avg"));
+        aggBuilder.subAggregation(new CumulativeSumPipelineAggregationBuilder("cusum", "the_deriv"));
+
+        executeTestCase(query, aggBuilder, histogram -> {
+            assertEquals(10, ((Histogram)histogram).getBuckets().size());
+            List<? extends Histogram.Bucket> buckets = ((Histogram)histogram).getBuckets();
+            double sum = 0.0;
+            for (int i = 0; i < buckets.size(); i++) {
+                if (i == 0) {
+                    assertThat(((InternalSimpleValue)(buckets.get(i).getAggregations().get("cusum"))).value(), equalTo(0.0));
+                } else {
+                    sum += 1.0;
+                    assertThat(((InternalSimpleValue)(buckets.get(i).getAggregations().get("cusum"))).value(), equalTo(sum));
+                }
+            }
+        });
+    }
+
+    public void testDocCount() throws IOException {
+        Query query = new MatchAllDocsQuery();
+
+        int numDocs = randomIntBetween(6, 20);
+        int interval = randomIntBetween(2, 5);
+
+        int minRandomValue = 0;
+        int maxRandomValue = 20;
+
+        int numValueBuckets = ((maxRandomValue - minRandomValue) / interval) + 1;
+        long[] valueCounts = new long[numValueBuckets];
+
+        HistogramAggregationBuilder aggBuilder = new HistogramAggregationBuilder("histo")
+            .field(VALUE_FIELD)
+            .interval(interval)
+            .extendedBounds(minRandomValue, maxRandomValue);
+        aggBuilder.subAggregation(new CumulativeSumPipelineAggregationBuilder("cusum", "_count"));
+
+        executeTestCase(query, aggBuilder, histogram -> {
+            List<? extends Histogram.Bucket> buckets = ((Histogram)histogram).getBuckets();
+
+            assertThat(buckets.size(), equalTo(numValueBuckets));
+
+            double sum = 0;
+            for (int i = 0; i < numValueBuckets; ++i) {
+                Histogram.Bucket bucket = buckets.get(i);
+                assertThat(bucket, notNullValue());
+                assertThat(((Number) bucket.getKey()).longValue(), equalTo((long) i * interval));
+                assertThat(bucket.getDocCount(), equalTo(valueCounts[i]));
+                sum += bucket.getDocCount();
+                InternalSimpleValue cumulativeSumValue = bucket.getAggregations().get("cusum");
+                assertThat(cumulativeSumValue, notNullValue());
+                assertThat(cumulativeSumValue.getName(), equalTo("cusum"));
+                assertThat(cumulativeSumValue.value(), equalTo(sum));
+            }
+        }, indexWriter -> {
+            Document document = new Document();
+
+            for (int i = 0; i < numDocs; i++) {
+                int fieldValue = randomIntBetween(minRandomValue, maxRandomValue);
+                document.add(new NumericDocValuesField(VALUE_FIELD, fieldValue));
+                final int bucket = (fieldValue / interval);
+                valueCounts[bucket]++;
+
+                indexWriter.addDocument(document);
+                document.clear();
+            }
+        });
+    }
+
+    public void testMetric() throws IOException {
+        Query query = new MatchAllDocsQuery();
+
+        int numDocs = randomIntBetween(6, 20);
+        int interval = randomIntBetween(2, 5);
+
+        int minRandomValue = 0;
+        int maxRandomValue = 20;
+
+        int numValueBuckets = ((maxRandomValue - minRandomValue) / interval) + 1;
+        long[] valueCounts = new long[numValueBuckets];
+
+        HistogramAggregationBuilder aggBuilder = new HistogramAggregationBuilder("histo")
+            .field(VALUE_FIELD)
+            .interval(interval)
+            .extendedBounds(minRandomValue, maxRandomValue);
+        aggBuilder.subAggregation(new SumAggregationBuilder("sum").field(VALUE_FIELD));
+        aggBuilder.subAggregation(new CumulativeSumPipelineAggregationBuilder("cusum", "sum"));
+
+        executeTestCase(query, aggBuilder, histogram -> {
+            List<? extends Histogram.Bucket> buckets = ((Histogram)histogram).getBuckets();
+
+            assertThat(buckets.size(), equalTo(numValueBuckets));
+
+            double bucketSum = 0;
+            for (int i = 0; i < numValueBuckets; ++i) {
+                Histogram.Bucket bucket = buckets.get(i);
+                assertThat(bucket, notNullValue());
+                assertThat(((Number) bucket.getKey()).longValue(), equalTo((long) i * interval));
+                Sum sum = bucket.getAggregations().get("sum");
+                assertThat(sum, notNullValue());
+                bucketSum += sum.value();
+
+                InternalSimpleValue sumBucketValue = bucket.getAggregations().get("cusum");
+                assertThat(sumBucketValue, notNullValue());
+                assertThat(sumBucketValue.getName(), equalTo("cusum"));
+                assertThat(sumBucketValue.value(), equalTo(bucketSum));
+            }
+        }, indexWriter -> {
+            Document document = new Document();
+
+            for (int i = 0; i < numDocs; i++) {
+                int fieldValue = randomIntBetween(minRandomValue, maxRandomValue);
+                document.add(new NumericDocValuesField(VALUE_FIELD, fieldValue));
+                final int bucket = (fieldValue / interval);
+                valueCounts[bucket]++;
+
+                indexWriter.addDocument(document);
+                document.clear();
+            }
+        });
+    }
+
+    public void testNoBuckets() throws IOException {
+        int numDocs = randomIntBetween(6, 20);
+        int interval = randomIntBetween(2, 5);
+
+        int minRandomValue = 0;
+        int maxRandomValue = 20;
+
+        int numValueBuckets = ((maxRandomValue - minRandomValue) / interval) + 1;
+        long[] valueCounts = new long[numValueBuckets];
+
+        Query query = new MatchNoDocsQuery();
+
+        HistogramAggregationBuilder aggBuilder = new HistogramAggregationBuilder("histo")
+            .field(VALUE_FIELD)
+            .interval(interval);
+        aggBuilder.subAggregation(new SumAggregationBuilder("sum").field(VALUE_FIELD));
+        aggBuilder.subAggregation(new CumulativeSumPipelineAggregationBuilder("cusum", "sum"));
+
+        executeTestCase(query, aggBuilder, histogram -> {
+            List<? extends Histogram.Bucket> buckets = ((Histogram)histogram).getBuckets();
+
+            assertThat(buckets.size(), equalTo(0));
+
+        }, indexWriter -> {
+            Document document = new Document();
+
+            for (int i = 0; i < numDocs; i++) {
+                int fieldValue = randomIntBetween(minRandomValue, maxRandomValue);
+                document.add(new NumericDocValuesField(VALUE_FIELD, fieldValue));
+                final int bucket = (fieldValue / interval);
+                valueCounts[bucket]++;
+
+                indexWriter.addDocument(document);
+                document.clear();
+            }
+        });
+    }
+
+    @SuppressWarnings("unchecked")
+    private void executeTestCase(Query query, AggregationBuilder aggBuilder, Consumer<InternalAggregation> verify) throws IOException {
+        executeTestCase(query, aggBuilder, verify, indexWriter -> {
+            Document document = new Document();
+            int counter = 0;
+            for (String date : datasetTimes) {
+                if (frequently()) {
+                    indexWriter.commit();
+                }
+
+                long instant = asLong(date);
+                document.add(new SortedNumericDocValuesField(HISTO_FIELD, instant));
+                document.add(new NumericDocValuesField(VALUE_FIELD, datasetValues.get(counter)));
+                indexWriter.addDocument(document);
+                document.clear();
+                counter += 1;
+            }
+        });
+    }
+
+    @SuppressWarnings("unchecked")
+    private void executeTestCase(Query query, AggregationBuilder aggBuilder, Consumer<InternalAggregation> verify,
+                                 CheckedConsumer<RandomIndexWriter, IOException> setup) throws IOException {
+
+        try (Directory directory = newDirectory()) {
+            try (RandomIndexWriter indexWriter = new RandomIndexWriter(random(), directory)) {
+               setup.accept(indexWriter);
+            }
+
+            try (IndexReader indexReader = DirectoryReader.open(directory)) {
+                IndexSearcher indexSearcher = newSearcher(indexReader, true, true);
+
+                DateFieldMapper.Builder builder = new DateFieldMapper.Builder("_name");
+                DateFieldMapper.DateFieldType fieldType = builder.fieldType();
+                fieldType.setHasDocValues(true);
+                fieldType.setName(HISTO_FIELD);
+
+                MappedFieldType valueFieldType = new NumberFieldMapper.NumberFieldType(NumberFieldMapper.NumberType.LONG);
+                valueFieldType.setHasDocValues(true);
+                valueFieldType.setName("value_field");
+
+                InternalAggregation histogram;
+                histogram = searchAndReduce(indexSearcher, query, aggBuilder, new MappedFieldType[]{fieldType, valueFieldType});
+                verify.accept(histogram);
+            }
+        }
+    }
+
+    private static long asLong(String dateTime) {
+        return DateFieldMapper.DEFAULT_DATE_TIME_FORMATTER.parser().parseDateTime(dateTime).getMillis();
+    }
+}