Collect and pass unparsed params to buildFactory(), replacing doParse()

doParse() was supposed to allow aggs to perform extra parsing.  Unfortunately, this forced the
parser to carry instance-level state, which would carry-over and "corrupt" any other aggs of the
same type in the same query.

Instead, we are now collecting all unknown params and pasing them as a Map<String, Object>
to buildFactory().  The agg may then parse them and instantiate a factory.  Each param the
agg uses, it should unset from the unusedParams object.

After building the factory, the parser verifies that unusedParams is empty.  If it is not empty,
an exception is raised so the user knows they provided unknown params.

Fixes #13337

core/src/main/java/org/elasticsearch/search/aggregations/pipeline/bucketmetrics/BucketMetricsParser.java

@@ -31,8 +31,11 @@ import org.elasticsearch.search.aggregations.support.format.ValueFormatter;
 import org.elasticsearch.search.internal.SearchContext;
 
 import java.io.IOException;
+import java.text.ParseException;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 /**
  * A parser for parsing requests for a {@link BucketMetricsPipelineAggregator}
@@ -52,12 +55,11 @@ public abstract class BucketMetricsParser implements PipelineAggregator.Parser {
         String[] bucketsPaths = null;
         String format = null;
         GapPolicy gapPolicy = GapPolicy.SKIP;
+        Map<String, Object> leftover = new HashMap<>(5);
 
         while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
             if (token == XContentParser.Token.FIELD_NAME) {
                 currentFieldName = parser.currentName();
-            } else if (doParse(pipelineAggregatorName, currentFieldName, token, parser, context)) {
-                // Do nothing as subclass has stored the state for this token
             } else if (token == XContentParser.Token.VALUE_STRING) {
                 if (context.parseFieldMatcher().match(currentFieldName, FORMAT)) {
                     format = parser.text();
@@ -66,8 +68,7 @@ public abstract class BucketMetricsParser implements PipelineAggregator.Parser {
                 } else if (context.parseFieldMatcher().match(currentFieldName, GAP_POLICY)) {
                     gapPolicy = GapPolicy.parse(context, parser.text(), parser.getTokenLocation());
                 } else {
-                    throw new SearchParseException(context, "Unknown key for a " + token + " in [" + pipelineAggregatorName + "]: ["
+                    leftover.put(currentFieldName, parser.text());
-                            + currentFieldName + "].", parser.getTokenLocation());
                 }
             } else if (token == XContentParser.Token.START_ARRAY) {
                 if (context.parseFieldMatcher().match(currentFieldName, BUCKETS_PATH)) {
@@ -78,18 +79,16 @@ public abstract class BucketMetricsParser implements PipelineAggregator.Parser {
                     }
                     bucketsPaths = paths.toArray(new String[paths.size()]);
                 } else {
-                    throw new SearchParseException(context, "Unknown key for a " + token + " in [" + pipelineAggregatorName + "]: ["
+                    leftover.put(currentFieldName, parser.list());
-                            + currentFieldName + "].", parser.getTokenLocation());
                 }
             } else {
-                throw new SearchParseException(context, "Unexpected token " + token + " in [" + pipelineAggregatorName + "].",
+                leftover.put(currentFieldName, parser.objectText());
-                        parser.getTokenLocation());
             }
         }
 
         if (bucketsPaths == null) {
             throw new SearchParseException(context, "Missing required field [" + BUCKETS_PATH.getPreferredName()
-                    + "] for derivative aggregation [" + pipelineAggregatorName + "]", parser.getTokenLocation());
+                    + "] for aggregation [" + pipelineAggregatorName + "]", parser.getTokenLocation());
         }
 
         ValueFormatter formatter = null;
@@ -99,15 +98,23 @@ public abstract class BucketMetricsParser implements PipelineAggregator.Parser {
             formatter = ValueFormatter.RAW;
         }
 
-        return buildFactory(pipelineAggregatorName, bucketsPaths, gapPolicy, formatter);
+        PipelineAggregatorFactory factory = null;
+        try {
+            factory = buildFactory(pipelineAggregatorName, bucketsPaths, gapPolicy, formatter, leftover);
+        } catch (ParseException exception) {
+            throw new SearchParseException(context, "Could not parse settings for aggregation ["
+                    + pipelineAggregatorName + "].", null, exception);
+        }
+
+        if (leftover.size() > 0) {
+            throw new SearchParseException(context, "Unexpected tokens " + leftover.keySet() + " in [" + pipelineAggregatorName + "].", null);
+        }
+        assert(factory != null);
+
+        return factory;
     }
 
     protected abstract PipelineAggregatorFactory buildFactory(String pipelineAggregatorName, String[] bucketsPaths, GapPolicy gapPolicy,
-            ValueFormatter formatter);
+            ValueFormatter formatter, Map<String, Object> unparsedParams) throws ParseException;
-
-    protected boolean doParse(String pipelineAggregatorName, String currentFieldName, Token token,
-                              XContentParser parser, SearchContext context) throws IOException {
-        return false;
-    }
 
 }

core/src/main/java/org/elasticsearch/search/aggregations/pipeline/bucketmetrics/avg/AvgBucketParser.java

@@ -24,6 +24,8 @@ import org.elasticsearch.search.aggregations.pipeline.PipelineAggregatorFactory;
 import org.elasticsearch.search.aggregations.pipeline.bucketmetrics.BucketMetricsParser;
 import org.elasticsearch.search.aggregations.support.format.ValueFormatter;
 
+import java.util.Map;
+
 public class AvgBucketParser extends BucketMetricsParser {
     @Override
     public String type() {
@@ -32,7 +34,7 @@ public class AvgBucketParser extends BucketMetricsParser {
 
     @Override
     protected PipelineAggregatorFactory buildFactory(String pipelineAggregatorName, String[] bucketsPaths, GapPolicy gapPolicy,
-            ValueFormatter formatter) {
+            ValueFormatter formatter, Map<String, Object> unparsedParams) {
         return new AvgBucketPipelineAggregator.Factory(pipelineAggregatorName, bucketsPaths, gapPolicy, formatter);
     }
 }

core/src/main/java/org/elasticsearch/search/aggregations/pipeline/bucketmetrics/max/MaxBucketParser.java

@@ -24,6 +24,8 @@ import org.elasticsearch.search.aggregations.pipeline.BucketHelpers.GapPolicy;
 import org.elasticsearch.search.aggregations.pipeline.bucketmetrics.BucketMetricsParser;
 import org.elasticsearch.search.aggregations.support.format.ValueFormatter;
 
+import java.util.Map;
+
 public class MaxBucketParser extends BucketMetricsParser {
 
     @Override
@@ -32,7 +34,8 @@ public class MaxBucketParser extends BucketMetricsParser {
     }
 
     @Override
-    protected PipelineAggregatorFactory buildFactory(String pipelineAggregatorName, String[] bucketsPaths, GapPolicy gapPolicy, ValueFormatter formatter) {
+    protected PipelineAggregatorFactory buildFactory(String pipelineAggregatorName, String[] bucketsPaths, GapPolicy gapPolicy,
+                                                     ValueFormatter formatter, Map<String, Object> unparsedParams) {
         return new MaxBucketPipelineAggregator.Factory(pipelineAggregatorName, bucketsPaths, gapPolicy, formatter);
     }
 

core/src/main/java/org/elasticsearch/search/aggregations/pipeline/bucketmetrics/min/MinBucketParser.java

@@ -24,6 +24,8 @@ import org.elasticsearch.search.aggregations.pipeline.PipelineAggregatorFactory;
 import org.elasticsearch.search.aggregations.pipeline.bucketmetrics.BucketMetricsParser;
 import org.elasticsearch.search.aggregations.support.format.ValueFormatter;
 
+import java.util.Map;
+
 public class MinBucketParser extends BucketMetricsParser {
 
     @Override
@@ -32,7 +34,7 @@ public class MinBucketParser extends BucketMetricsParser {
     }
 
     protected PipelineAggregatorFactory buildFactory(String pipelineAggregatorName, String[] bucketsPaths, GapPolicy gapPolicy,
-            ValueFormatter formatter) {
+            ValueFormatter formatter, Map<String, Object> unparsedParams) {
         return new MinBucketPipelineAggregator.Factory(pipelineAggregatorName, bucketsPaths, gapPolicy, formatter);
     };
 

core/src/main/java/org/elasticsearch/search/aggregations/pipeline/bucketmetrics/percentile/PercentilesBucketParser.java

@@ -19,17 +19,14 @@
 
 package org.elasticsearch.search.aggregations.pipeline.bucketmetrics.percentile;
 
-import com.google.common.primitives.Doubles;
 import org.elasticsearch.common.ParseField;
-import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.search.aggregations.pipeline.PipelineAggregatorFactory;
 import org.elasticsearch.search.aggregations.pipeline.bucketmetrics.BucketMetricsParser;
 import org.elasticsearch.search.aggregations.support.format.ValueFormatter;
-import org.elasticsearch.search.internal.SearchContext;
 
-import java.io.IOException;
+import java.text.ParseException;
-import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 
 import static org.elasticsearch.search.aggregations.pipeline.BucketHelpers.GapPolicy;
 
@@ -37,7 +34,6 @@ import static org.elasticsearch.search.aggregations.pipeline.BucketHelpers.GapPo
 public class PercentilesBucketParser extends BucketMetricsParser {
 
     public static final ParseField PERCENTS = new ParseField("percents");
-    double[] percents = new double[] { 1.0, 5.0, 25.0, 50.0, 75.0, 95.0, 99.0 };
 
     @Override
     public String type() {
@@ -46,22 +42,31 @@ public class PercentilesBucketParser extends BucketMetricsParser {
 
     @Override
     protected PipelineAggregatorFactory buildFactory(String pipelineAggregatorName, String[] bucketsPaths, GapPolicy gapPolicy,
-                                                     ValueFormatter formatter) {
+                                                     ValueFormatter formatter, Map<String, Object> unparsedParams) throws ParseException {
+
+        double[] percents = new double[] { 1.0, 5.0, 25.0, 50.0, 75.0, 95.0, 99.0 };
+        int counter = 0;
+        Object percentParam = unparsedParams.get(PERCENTS.getPreferredName());
+
+        if (percentParam != null) {
+            if (percentParam instanceof List) {
+                percents = new double[((List) percentParam).size()];
+                for (Object p : (List) percentParam) {
+                    if (p instanceof Double) {
+                        percents[counter] = (Double) p;
+                        counter += 1;
+                    } else {
+                        throw new ParseException("Parameter [" + PERCENTS.getPreferredName() + "] must be an array of doubles, type `"
+                                + percentParam.getClass().getSimpleName() + "` provided instead", 0);
+                    }
+                }
+                unparsedParams.remove(PERCENTS.getPreferredName());
+            } else {
+                throw new ParseException("Parameter [" + PERCENTS.getPreferredName() + "] must be an array of doubles, type `"
+                        + percentParam.getClass().getSimpleName() + "` provided instead", 0);
+            }
+        }
+
         return new PercentilesBucketPipelineAggregator.Factory(pipelineAggregatorName, bucketsPaths, gapPolicy, formatter, percents);
     }
-
-    @Override
-    protected boolean doParse(String pipelineAggregatorName, String currentFieldName,
-                              XContentParser.Token token, XContentParser parser, SearchContext context) throws IOException {
-        if (context.parseFieldMatcher().match(currentFieldName, PERCENTS)) {
-
-            List<Double> parsedPercents = new ArrayList<>();
-            while ((token = parser.nextToken()) != XContentParser.Token.END_ARRAY) {
-                parsedPercents.add(parser.doubleValue());
-            }
-            percents = Doubles.toArray(parsedPercents);
-            return true;
-        }
-        return false;
-    }
 }

core/src/main/java/org/elasticsearch/search/aggregations/pipeline/bucketmetrics/sum/SumBucketParser.java

@@ -24,6 +24,8 @@ import org.elasticsearch.search.aggregations.pipeline.PipelineAggregatorFactory;
 import org.elasticsearch.search.aggregations.pipeline.bucketmetrics.BucketMetricsParser;
 import org.elasticsearch.search.aggregations.support.format.ValueFormatter;
 
+import java.util.Map;
+
 public class SumBucketParser extends BucketMetricsParser {
     @Override
     public String type() {
@@ -32,7 +34,7 @@ public class SumBucketParser extends BucketMetricsParser {
 
     @Override
     protected PipelineAggregatorFactory buildFactory(String pipelineAggregatorName, String[] bucketsPaths, GapPolicy gapPolicy,
-            ValueFormatter formatter) {
+            ValueFormatter formatter, Map<String, Object> unparsedParams) {
         return new SumBucketPipelineAggregator.Factory(pipelineAggregatorName, bucketsPaths, gapPolicy, formatter);
     }
 }

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

@@ -489,7 +489,6 @@ public class PercentilesBucketIT extends ESIntegTestCase {
     }
 
     @Test
-    @AwaitsFix(bugUrl="https://github.com/elastic/elasticsearch/issues/13337")
     public void testNested() throws Exception {
         SearchResponse response = client()
                 .prepareSearch("idx")