Add pipeline parent validation for auto date histogram (#35670)

Allow `auto_date_histogram` as a valid parent agg for derivative, 
cumulative sum, moving average, moving function and serial 
differencing pipeline aggregations.

Since all these aggs share the same requirement (sequentially
ordered parent aggs), this commit also refactors to share
the same validation code so that any newly added aggs won't
be forgotten.

Closes #35578

Author: jklancic

server/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/HistogramAggregatorFactory.java

@@ -42,7 +42,7 @@ public final class HistogramAggregatorFactory extends ValuesSourceAggregatorFact
     private final long minDocCount;
     private final double minBound, maxBound;
 
-    HistogramAggregatorFactory(String name, ValuesSourceConfig<Numeric> config, double interval, double offset,
+    public HistogramAggregatorFactory(String name, ValuesSourceConfig<Numeric> config, double interval, double offset,
             BucketOrder order, boolean keyed, long minDocCount, double minBound, double maxBound,
             SearchContext context, AggregatorFactory<?> parent,
             AggregatorFactories.Builder subFactoriesBuilder, Map<String, Object> metaData) throws IOException {

server/src/main/java/org/elasticsearch/search/aggregations/pipeline/AbstractPipelineAggregationBuilder.java

@@ -25,6 +25,9 @@
 import org.elasticsearch.search.aggregations.AggregationBuilder;
 import org.elasticsearch.search.aggregations.AggregatorFactory;
 import org.elasticsearch.search.aggregations.PipelineAggregationBuilder;
+import org.elasticsearch.search.aggregations.bucket.histogram.AutoDateHistogramAggregatorFactory;
+import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramAggregatorFactory;
+import org.elasticsearch.search.aggregations.bucket.histogram.HistogramAggregatorFactory;
 
 import java.io.IOException;
 import java.util.Arrays;
@@ -102,6 +105,28 @@ public final PipelineAggregator create() throws IOException {
     public void doValidate(AggregatorFactory<?> parent, Collection<AggregationBuilder> factories,
             Collection<PipelineAggregationBuilder> pipelineAggregatorFactories) {
     }
+    
+    /**
+     * Validates pipeline aggregations that need sequentially ordered data.
+     */
+    public static void validateSequentiallyOrderedParentAggs(AggregatorFactory<?> parent, String type, String name) {
+        if ((parent instanceof HistogramAggregatorFactory || parent instanceof DateHistogramAggregatorFactory
+                || parent instanceof AutoDateHistogramAggregatorFactory) == false) {
+            throw new IllegalStateException(
+                    type + " aggregation [" + name + "] must have a histogram, date_histogram or auto_date_histogram as parent");
+        }
+        if (parent instanceof HistogramAggregatorFactory) {
+            HistogramAggregatorFactory histoParent = (HistogramAggregatorFactory) parent;
+            if (histoParent.minDocCount() != 0) {
+                throw new IllegalStateException("parent histogram of " + type + " aggregation [" + name + "] must have min_doc_count of 0");
+            }
+        } else if (parent instanceof DateHistogramAggregatorFactory) {
+            DateHistogramAggregatorFactory histoParent = (DateHistogramAggregatorFactory) parent;
+            if (histoParent.minDocCount() != 0) {
+                throw new IllegalStateException("parent histogram of " + type + " aggregation [" + name + "] must have min_doc_count of 0");
+            }
+        }
+    }
 
     @SuppressWarnings("unchecked")
     @Override

server/src/main/java/org/elasticsearch/search/aggregations/pipeline/CumulativeSumPipelineAggregationBuilder.java

@@ -28,8 +28,6 @@
 import org.elasticsearch.search.aggregations.AggregationBuilder;
 import org.elasticsearch.search.aggregations.AggregatorFactory;
 import org.elasticsearch.search.aggregations.PipelineAggregationBuilder;
-import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramAggregatorFactory;
-import org.elasticsearch.search.aggregations.bucket.histogram.HistogramAggregatorFactory;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -101,22 +99,8 @@ public void doValidate(AggregatorFactory<?> parent, Collection<AggregationBuilde
             throw new IllegalStateException(BUCKETS_PATH.getPreferredName()
                     + " must contain a single entry for aggregation [" + name + "]");
         }
-        if (parent instanceof HistogramAggregatorFactory) {
-            HistogramAggregatorFactory histoParent = (HistogramAggregatorFactory) parent;
-            if (histoParent.minDocCount() != 0) {
-                throw new IllegalStateException("parent histogram of cumulative sum aggregation [" + name
-                        + "] must have min_doc_count of 0");
-            }
-        } else if (parent instanceof DateHistogramAggregatorFactory) {
-            DateHistogramAggregatorFactory histoParent = (DateHistogramAggregatorFactory) parent;
-            if (histoParent.minDocCount() != 0) {
-                throw new IllegalStateException("parent histogram of cumulative sum aggregation [" + name
-                        + "] must have min_doc_count of 0");
-            }
-        } else {
-            throw new IllegalStateException("cumulative sum aggregation [" + name
-                    + "] must have a histogram or date_histogram as parent");
-        }
+        
+        validateSequentiallyOrderedParentAggs(parent, NAME, name);
     }
 
     @Override

server/src/main/java/org/elasticsearch/search/aggregations/pipeline/DerivativePipelineAggregationBuilder.java

@@ -32,9 +32,7 @@
 import org.elasticsearch.search.aggregations.AggregatorFactory;
 import org.elasticsearch.search.aggregations.PipelineAggregationBuilder;
 import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramAggregationBuilder;
-import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramAggregatorFactory;
 import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval;
-import org.elasticsearch.search.aggregations.bucket.histogram.HistogramAggregatorFactory;
 import org.elasticsearch.search.aggregations.pipeline.BucketHelpers.GapPolicy;
 import org.joda.time.DateTimeZone;
 
@@ -161,22 +159,8 @@ public void doValidate(AggregatorFactory<?> parent, Collection<AggregationBuilde
             throw new IllegalStateException(PipelineAggregator.Parser.BUCKETS_PATH.getPreferredName()
                     + " must contain a single entry for aggregation [" + name + "]");
         }
-        if (parent instanceof HistogramAggregatorFactory) {
-            HistogramAggregatorFactory histoParent = (HistogramAggregatorFactory) parent;
-            if (histoParent.minDocCount() != 0) {
-                throw new IllegalStateException("parent histogram of derivative aggregation [" + name
-                        + "] must have min_doc_count of 0");
-            }
-        } else if (parent instanceof DateHistogramAggregatorFactory) {
-            DateHistogramAggregatorFactory histoParent = (DateHistogramAggregatorFactory) parent;
-            if (histoParent.minDocCount() != 0) {
-                throw new IllegalStateException("parent histogram of derivative aggregation [" + name
-                        + "] must have min_doc_count of 0");
-            }
-        } else {
-            throw new IllegalStateException("derivative aggregation [" + name
-                    + "] must have a histogram or date_histogram as parent");
-        }
+        
+        validateSequentiallyOrderedParentAggs(parent, NAME, name);
     }
 
     @Override

server/src/main/java/org/elasticsearch/search/aggregations/pipeline/MovAvgPipelineAggregationBuilder.java

@@ -32,8 +32,6 @@
 import org.elasticsearch.search.aggregations.AggregationBuilder;
 import org.elasticsearch.search.aggregations.AggregatorFactory;
 import org.elasticsearch.search.aggregations.PipelineAggregationBuilder;
-import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramAggregatorFactory;
-import org.elasticsearch.search.aggregations.bucket.histogram.HistogramAggregatorFactory;
 import org.elasticsearch.search.aggregations.pipeline.BucketHelpers.GapPolicy;
 
 import java.io.IOException;
@@ -267,22 +265,8 @@ public void doValidate(AggregatorFactory<?> parent, Collection<AggregationBuilde
             throw new IllegalStateException(PipelineAggregator.Parser.BUCKETS_PATH.getPreferredName()
                     + " must contain a single entry for aggregation [" + name + "]");
         }
-        if (parent instanceof HistogramAggregatorFactory) {
-            HistogramAggregatorFactory histoParent = (HistogramAggregatorFactory) parent;
-            if (histoParent.minDocCount() != 0) {
-                throw new IllegalStateException("parent histogram of moving average aggregation [" + name
-                        + "] must have min_doc_count of 0");
-            }
-        } else if (parent instanceof DateHistogramAggregatorFactory) {
-            DateHistogramAggregatorFactory histoParent = (DateHistogramAggregatorFactory) parent;
-            if (histoParent.minDocCount() != 0) {
-                throw new IllegalStateException("parent histogram of moving average aggregation [" + name
-                        + "] must have min_doc_count of 0");
-            }
-        } else {
-            throw new IllegalStateException("moving average aggregation [" + name
-                    + "] must have a histogram or date_histogram as parent");
-        }
+        
+        validateSequentiallyOrderedParentAggs(parent, NAME, name);
     }
 
     @Override

server/src/main/java/org/elasticsearch/search/aggregations/pipeline/MovFnPipelineAggregationBuilder.java

@@ -32,8 +32,6 @@
 import org.elasticsearch.search.aggregations.AggregationBuilder;
 import org.elasticsearch.search.aggregations.AggregatorFactory;
 import org.elasticsearch.search.aggregations.PipelineAggregationBuilder;
-import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramAggregatorFactory;
-import org.elasticsearch.search.aggregations.bucket.histogram.HistogramAggregatorFactory;
 import org.elasticsearch.search.aggregations.pipeline.BucketHelpers.GapPolicy;
 
 import java.io.IOException;
@@ -176,22 +174,8 @@ public void doValidate(AggregatorFactory<?> parent, Collection<AggregationBuilde
         if (window <= 0) {
             throw new IllegalArgumentException("[" + WINDOW.getPreferredName() + "] must be a positive, non-zero integer.");
         }
-        if (parent instanceof HistogramAggregatorFactory) {
-            HistogramAggregatorFactory histoParent = (HistogramAggregatorFactory) parent;
-            if (histoParent.minDocCount() != 0) {
-                throw new IllegalStateException("parent histogram of moving_function aggregation [" + name
-                    + "] must have min_doc_count of 0");
-            }
-        } else if (parent instanceof DateHistogramAggregatorFactory) {
-            DateHistogramAggregatorFactory histoParent = (DateHistogramAggregatorFactory) parent;
-            if (histoParent.minDocCount() != 0) {
-                throw new IllegalStateException("parent histogram of moving_function aggregation [" + name
-                    + "] must have min_doc_count of 0");
-            }
-        } else {
-            throw new IllegalStateException("moving_function aggregation [" + name
-                + "] must have a histogram or date_histogram as parent");
-        }
+        
+        validateSequentiallyOrderedParentAggs(parent, NAME, name);
     }
 
     @Override

server/src/main/java/org/elasticsearch/search/aggregations/pipeline/SerialDiffPipelineAggregationBuilder.java

@@ -26,10 +26,14 @@
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.search.DocValueFormat;
+import org.elasticsearch.search.aggregations.AggregationBuilder;
+import org.elasticsearch.search.aggregations.AggregatorFactory;
+import org.elasticsearch.search.aggregations.PipelineAggregationBuilder;
 import org.elasticsearch.search.aggregations.pipeline.BucketHelpers.GapPolicy;
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -134,6 +138,12 @@ protected DocValueFormat formatter() {
     protected PipelineAggregator createInternal(Map<String, Object> metaData) throws IOException {
         return new SerialDiffPipelineAggregator(name, bucketsPaths, formatter(), gapPolicy, lag, metaData);
     }
+    
+    @Override
+    public void doValidate(AggregatorFactory<?> parent, Collection<AggregationBuilder> aggFactories,
+            Collection<PipelineAggregationBuilder> pipelineAggregatoractories) {        
+        validateSequentiallyOrderedParentAggs(parent, NAME, name);
+    }
 
     @Override
     protected XContentBuilder internalXContent(XContentBuilder builder, Params params) throws IOException {

server/src/test/java/org/elasticsearch/search/aggregations/MultiBucketAggregatorWrapperTests.java

@@ -30,17 +30,14 @@
 import org.elasticsearch.search.internal.SearchContext;
 import org.elasticsearch.test.ESTestCase;
 
-import java.io.IOException;
 import java.util.Collections;
-import java.util.List;
-import java.util.Map;
 
+import static org.mockito.Matchers.same;
 import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.reset;
-import static org.mockito.Mockito.same;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
 
 public class MultiBucketAggregatorWrapperTests extends ESTestCase {
 
@@ -74,21 +71,4 @@ public void testNoNullScorerIsDelegated() throws Exception {
         verifyNoMoreInteractions(wrappedCollector);
         wrapper.close();
     }
-
-    static class TestAggregatorFactory extends AggregatorFactory {
-
-        private final Aggregator aggregator;
-
-        TestAggregatorFactory(SearchContext context, Aggregator aggregator) throws IOException {
-            super("_name", context, null, new AggregatorFactories.Builder(), Collections.emptyMap());
-            this.aggregator = aggregator;
-        }
-
-        @Override
-        protected Aggregator createInternal(Aggregator parent, boolean collectsFromSingleBucket, List list,
-                                            Map metaData) throws IOException {
-            return aggregator;
-        }
-    }
-
 }

server/src/test/java/org/elasticsearch/search/aggregations/TestAggregatorFactory.java

@@ -0,0 +1,64 @@
+/*
+ * 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;
+
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.BigArrays;
+import org.elasticsearch.common.util.MockBigArrays;
+import org.elasticsearch.common.util.MockPageCacheRecycler;
+import org.elasticsearch.indices.breaker.NoneCircuitBreakerService;
+import org.elasticsearch.search.internal.SearchContext;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Test implementation for AggregatorFactory.
+ */
+public class TestAggregatorFactory extends AggregatorFactory {
+
+    private final Aggregator aggregator;
+
+    TestAggregatorFactory(SearchContext context, Aggregator aggregator) throws IOException {
+        super("_name", context, null, new AggregatorFactories.Builder(), Collections.emptyMap());
+        this.aggregator = aggregator;
+    }
+
+    @Override
+    protected Aggregator createInternal(Aggregator parent, boolean collectsFromSingleBucket, List list,
+                                        Map metaData) throws IOException {
+        return aggregator;
+    }
+    
+    public static TestAggregatorFactory createInstance() throws IOException {
+        BigArrays bigArrays = new MockBigArrays(new MockPageCacheRecycler(Settings.EMPTY), new NoneCircuitBreakerService());
+        SearchContext searchContext = mock(SearchContext.class);
+        when(searchContext.bigArrays()).thenReturn(bigArrays);
+
+        Aggregator aggregator = mock(Aggregator.class);
+        
+        return new TestAggregatorFactory(searchContext, aggregator);
+    }
+}

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

@@ -37,6 +37,8 @@
 import org.elasticsearch.search.aggregations.AggregationBuilder;
 import org.elasticsearch.search.aggregations.AggregatorTestCase;
 import org.elasticsearch.search.aggregations.InternalAggregation;
+import org.elasticsearch.search.aggregations.PipelineAggregationBuilder;
+import org.elasticsearch.search.aggregations.TestAggregatorFactory;
 import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramAggregationBuilder;
 import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval;
 import org.elasticsearch.search.aggregations.bucket.histogram.Histogram;
@@ -48,7 +50,10 @@
 
 import java.io.IOException;
 import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
 import java.util.function.Consumer;
 
 import static org.hamcrest.Matchers.equalTo;
@@ -259,6 +264,34 @@ public void testNoBuckets() throws IOException {
             }
         });
     }
+    
+    /**
+     * The validation should verify the parent aggregation is allowed.
+     */
+    public void testValidate() throws IOException {
+        final Set<PipelineAggregationBuilder> aggBuilders = new HashSet<>();
+        aggBuilders.add(new CumulativeSumPipelineAggregationBuilder("cusum", "sum"));
+
+        final CumulativeSumPipelineAggregationBuilder builder = new CumulativeSumPipelineAggregationBuilder("name", "valid");
+        builder.validate(PipelineAggregationHelperTests.getRandomSequentiallyOrderedParentAgg(), Collections.emptySet(), aggBuilders);
+    }
+
+    /**
+     * The validation should throw an IllegalArgumentException, since parent
+     * aggregation is not a type of HistogramAggregatorFactory,
+     * DateHistogramAggregatorFactory or AutoDateHistogramAggregatorFactory.
+     */
+    public void testValidateException() throws IOException {
+        final Set<PipelineAggregationBuilder> aggBuilders = new HashSet<>();
+        aggBuilders.add(new CumulativeSumPipelineAggregationBuilder("cusum", "sum"));
+        TestAggregatorFactory parentFactory = TestAggregatorFactory.createInstance();
+
+        final CumulativeSumPipelineAggregationBuilder builder = new CumulativeSumPipelineAggregationBuilder("name", "invalid_agg>metric");
+        IllegalStateException ex = expectThrows(IllegalStateException.class,
+                () -> builder.validate(parentFactory, Collections.emptySet(), aggBuilders));
+        assertEquals("cumulative_sum aggregation [name] must have a histogram, date_histogram or auto_date_histogram as parent",
+                ex.getMessage());
+    }
 
     private void executeTestCase(Query query, AggregationBuilder aggBuilder, Consumer<InternalAggregation> verify) throws IOException {
         executeTestCase(query, aggBuilder, verify, indexWriter -> {