SQL: Add method args to PERCENTILE/PERCENTILE_RANK (#65026)

* Adds the capability to have functions with two optional arguments
* Adds two new optional arguments to `PERCENTILE()` and
  `PERCENTILE_RANK()` functions, namely the method and
  method_parameter which can be: 1) `tdigest` and a double `compression`
  parameter or 2) `hdr` and an integer representing the
  `number_of_digits` parameter.
* Integration tests
* Documentation updates

Closes #63567

docs/reference/sql/functions/aggs.asciidoc

@@ -477,14 +477,18 @@ include-tagged::{sql-specs}/docs/docs.csv-spec[aggMadScalars]
 [source, sql]
 --------------------------------------------------
 PERCENTILE(
-    field_name,  <1>
+    field_name,         <1>
-    numeric_exp) <2>
+    percentile[,        <2>
+    method[,            <3>
+    method_parameter]]) <4>
 --------------------------------------------------
 
 *Input*:
 
 <1> a numeric field
 <2> a numeric expression (must be a constant and not based on a field)
+<3> optional string literal for the <<search-aggregations-metrics-percentile-aggregation-approximation,percentile algorithm>>. Possible values: `tdigest` or `hdr`. Defaults to `tdigest`.
+<4> optional numeric literal that configures the <<search-aggregations-metrics-percentile-aggregation-approximation,percentile algorithm>>. Configures `compression` for `tdigest` or `number_of_significant_value_digits` for `hdr`. The default is the same as that of the backing algorithm.
 
 *Output*: `double` numeric value
 
@@ -503,6 +507,11 @@ include-tagged::{sql-specs}/docs/docs.csv-spec[aggPercentile]
 include-tagged::{sql-specs}/docs/docs.csv-spec[aggPercentileScalars]
 --------------------------------------------------
 
+["source","sql",subs="attributes,macros"]
+--------------------------------------------------
+include-tagged::{sql-specs}/docs/docs.csv-spec[aggPercentileWithPercentileConfig]
+--------------------------------------------------
+
 [[sql-functions-aggs-percentile-rank]]
 ==== `PERCENTILE_RANK`
 
@@ -510,14 +519,19 @@ include-tagged::{sql-specs}/docs/docs.csv-spec[aggPercentileScalars]
 [source, sql]
 --------------------------------------------------
 PERCENTILE_RANK(
-    field_name,  <1>
+    field_name,         <1>
-    numeric_exp) <2>
+    value[,             <2>
+    method[,            <3>
+    method_parameter]]) <4>
 --------------------------------------------------
 
 *Input*:
 
 <1> a numeric field
 <2> a numeric expression (must be a constant and not based on a field)
+<3> optional string literal for the <<search-aggregations-metrics-percentile-aggregation-approximation,percentile algorithm>>. Possible values: `tdigest` or `hdr`. Defaults to `tdigest`.
+<4> optional numeric literal that configures the <<search-aggregations-metrics-percentile-aggregation-approximation,percentile algorithm>>. Configures `compression` for `tdigest` or `number_of_significant_value_digits` for `hdr`. The default is the same as that of the backing algorithm.
+
 
 *Output*: `double` numeric value
 
@@ -536,6 +550,11 @@ include-tagged::{sql-specs}/docs/docs.csv-spec[aggPercentileRank]
 include-tagged::{sql-specs}/docs/docs.csv-spec[aggPercentileRankScalars]
 --------------------------------------------------
 
+["source","sql",subs="attributes,macros"]
+--------------------------------------------------
+include-tagged::{sql-specs}/docs/docs.csv-spec[aggPercentileRankWithPercentileConfig]
+--------------------------------------------------
+
 [[sql-functions-aggs-skewness]]
 ==== `SKEWNESS`
 

x-pack/plugin/ql/src/main/java/org/elasticsearch/xpack/ql/expression/function/FunctionRegistry.java

@@ -420,16 +420,23 @@ public class FunctionRegistry {
     public static <T extends Function> FunctionDefinition def(Class<T> function,
             FourParametersFunctionBuilder<T> ctorRef, String... names) {
         FunctionBuilder builder = (source, children, distinct, cfg) -> {
-            boolean hasMinimumThree = OptionalArgument.class.isAssignableFrom(function);
+            if (OptionalArgument.class.isAssignableFrom(function)) {
-            if (hasMinimumThree && (children.size() > 4 || children.size() < 3)) {
+                if (children.size() > 4 || children.size() < 3) {
-                throw new QlIllegalArgumentException("expects three or four arguments");
+                    throw new QlIllegalArgumentException("expects three or four arguments");
-            } else if (!hasMinimumThree && children.size() != 4) {
+                }
+            } else if (TwoOptionalArguments.class.isAssignableFrom(function)) {
+                if (children.size() > 4 || children.size() < 2) {
+                    throw new QlIllegalArgumentException("expects minimum two, maximum four arguments");
+                }
+            } else if (children.size() != 4) {
                 throw new QlIllegalArgumentException("expects exactly four arguments");
             }
             if (distinct) {
                 throw new QlIllegalArgumentException("does not support DISTINCT yet it was specified");
             }
-            return ctorRef.build(source, children.get(0), children.get(1), children.get(2), children.size() == 4 ? children.get(3) : null);
+            return ctorRef.build(source, children.get(0), children.get(1), 
+                children.size() > 2 ? children.get(2) : null,
+                children.size() > 3 ? children.get(3) : null);
         };
         return def(function, builder, false, names);
     }

x-pack/plugin/ql/src/main/java/org/elasticsearch/xpack/ql/expression/function/TwoOptionalArguments.java

@@ -0,0 +1,15 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.ql.expression.function;
+
+/**
+ * Marker interface indicating that a function accepts two optional arguments (the last two).
+ * This is used by the {@link FunctionRegistry} to perform validation of function declaration.
+ */
+public interface TwoOptionalArguments {
+
+}

x-pack/plugin/sql/qa/server/src/main/resources/agg.csv-spec

@@ -22,6 +22,42 @@ F                    |10099.7608
 M                    |10096.2232
 ;
 
+singlePercentileWithCommaTDigestSpecified
+SELECT gender, PERCENTILE(emp_no, 97.76, 'tdigest') p1 FROM test_emp GROUP BY gender;
+
+gender:s             | p1:d
+null                 |10019.0
+F                    |10099.7608
+M                    |10096.2232
+;
+
+singlePercentileWithCommaTDigestWithCompressionSpecified
+SELECT gender, PERCENTILE(emp_no, 97.76, 'tdigest', 50 + 0.2) p1 FROM test_emp GROUP BY gender;
+
+gender:s             | p1:d
+null                 |10019.0
+F                    |10099.7608
+M                    |10096.2232
+;
+
+singlePercentileWithCommaHDRSpecified
+SELECT gender, PERCENTILE(emp_no, 97.76, 'hdr') p1 FROM test_emp GROUP BY gender;
+
+gender:s             | p1:d
+null                 |10016.0
+F                    |10096.0
+M                    |10096.0
+;
+
+singlePercentileWithCommaHDRWithDigitsSpecified
+SELECT gender, PERCENTILE(emp_no, 97.76, 'hdr', 1+1) p1 FROM test_emp GROUP BY gender;
+
+gender:s             | p1:d
+null                 |9984.0
+F                    |10048.0
+M                    |10048.0
+;
+
 multiplePercentilesOneWithCommaOneWithout
 SELECT gender, PERCENTILE(emp_no, 92.45) p1, PERCENTILE(emp_no, 91) p2 FROM test_emp GROUP BY gender;
 
@@ -58,6 +94,24 @@ F                    |17.424242424242426
 M                    |15.350877192982457
 ;
 
+singlePercentileRankWithHDRSpecified
+SELECT gender, PERCENTILE_RANK(emp_no, 10000 + 25, 'hdr') p1 FROM test_emp GROUP BY gender;
+
+gender:s             | p1:d
+null                 |100.0
+F                    |21.21212121212121
+M                    |24.56140350877193
+;
+
+singlePercentileRankHDRWithDigitsSpecified
+SELECT gender, PERCENTILE_RANK(emp_no, 10000 + 25, 'hdr', 4+1) p1 FROM test_emp GROUP BY gender;
+
+gender:s             | p1:d
+null                 |100.0
+F                    |18.181818181818183
+M                    |15.789473684210526
+;
+
 multiplePercentileRanks
 SELECT gender, PERCENTILE_RANK(emp_no, 10030.0) rank1, PERCENTILE_RANK(emp_no, 10025) rank2 FROM test_emp GROUP BY gender;
 

x-pack/plugin/sql/qa/server/src/main/resources/docs/docs.csv-spec

@@ -1510,6 +1510,26 @@ null           |6249.916666666667
 // end::aggPercentileScalars
 ;
 
+aggPercentileWithPercentileConfig
+// tag::aggPercentileWithPercentileConfig
+SELECT
+    languages, 
+    PERCENTILE(salary, 97.3, 'tdigest', 100.0) AS "97.3_TDigest", 
+    PERCENTILE(salary, 97.3, 'hdr', 3) AS "97.3_HDR" 
+FROM emp 
+GROUP BY languages;
+
+   languages   | 97.3_TDigest  |   97.3_HDR    
+---------------+---------------+---------------
+null           |74999.0        |74992.0        
+1              |73717.0        |73712.0        
+2              |73530.238      |69936.0        
+3              |74970.0        |74992.0        
+4              |74572.0        |74608.0        
+5              |66117.118      |56368.0        
+// end::aggPercentileWithPercentileConfig
+;
+
 aggPercentileRank
 // tag::aggPercentileRank
 SELECT languages, PERCENTILE_RANK(salary, 65000) AS rank FROM emp GROUP BY languages;
@@ -1541,6 +1561,26 @@ null           |66.91240875912409
 // end::aggPercentileRankScalars
 ;
 
+aggPercentileRankWithPercentileConfig
+// tag::aggPercentileRankWithPercentileConfig
+SELECT
+    languages,
+    ROUND(PERCENTILE_RANK(salary, 65000, 'tdigest', 100.0), 2) AS "rank_TDigest",
+    ROUND(PERCENTILE_RANK(salary, 65000, 'hdr', 3), 2) AS "rank_HDR"
+FROM emp
+GROUP BY languages;
+
+   languages   | rank_TDigest  |   rank_HDR    
+---------------+---------------+---------------
+null           |73.66          |80.0           
+1              |73.73          |73.33          
+2              |88.88          |89.47          
+3              |79.44          |76.47          
+4              |85.7           |83.33          
+5              |100.0          |95.24          
+// end::aggPercentileRankWithPercentileConfig
+;
+
 aggSkewness
 // tag::aggSkewness
 SELECT MIN(salary) AS min, MAX(salary) AS max, SKEWNESS(salary) AS s FROM emp;

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/aggregate/Percentile.java

@@ -6,33 +6,20 @@
 package org.elasticsearch.xpack.sql.expression.function.aggregate;
 
 import org.elasticsearch.xpack.ql.expression.Expression;
-import org.elasticsearch.xpack.ql.expression.Expressions.ParamOrdinal;
-import org.elasticsearch.xpack.ql.expression.Foldables;
-import org.elasticsearch.xpack.ql.expression.function.aggregate.EnclosedAgg;
 import org.elasticsearch.xpack.ql.tree.NodeInfo;
 import org.elasticsearch.xpack.ql.tree.Source;
-import org.elasticsearch.xpack.ql.type.DataType;
-import org.elasticsearch.xpack.ql.type.DataTypes;
-import org.elasticsearch.xpack.sql.type.SqlDataTypeConverter;
 
 import java.util.List;
 
-import static java.util.Collections.singletonList;
+public class Percentile extends PercentileAggregate {
-import static org.elasticsearch.xpack.ql.expression.TypeResolutions.isFoldable;
-import static org.elasticsearch.xpack.ql.expression.TypeResolutions.isNumeric;
 
-public class Percentile extends NumericAggregate implements EnclosedAgg {
+    public Percentile(Source source, Expression field, Expression percent, Expression method, Expression methodParameter) {
-
+        super(source, field, percent, method, methodParameter);
-    private final Expression percent;
-
-    public Percentile(Source source, Expression field, Expression percent) {
-        super(source, field, singletonList(percent));
-        this.percent = percent;
     }
 
     @Override
     protected NodeInfo<Percentile> info() {
-        return NodeInfo.create(this, Percentile::new, field(), percent);
+        return NodeInfo.create(this, Percentile::new, field(), percent(), method(), methodParameter());
     }
 
     @Override
@@ -40,36 +27,10 @@ public class Percentile extends NumericAggregate implements EnclosedAgg {
         if (newChildren.size() != 2) {
             throw new IllegalArgumentException("expected [2] children but received [" + newChildren.size() + "]");
         }
-        return new Percentile(source(), newChildren.get(0), newChildren.get(1));
+        return new Percentile(source(), newChildren.get(0), newChildren.get(1), method(), methodParameter());
-    }
-
-    @Override
-    protected TypeResolution resolveType() {
-        TypeResolution resolution = isFoldable(percent, sourceText(), ParamOrdinal.SECOND);
-        if (resolution.unresolved()) {
-            return resolution;
-        }
-
-        resolution = super.resolveType();
-        if (resolution.unresolved()) {
-            return resolution;
-        }
-
-        return isNumeric(percent, sourceText(), ParamOrdinal.DEFAULT);
     }
 
     public Expression percent() {
-        return percent;
+        return parameter();
-    }
-
-    @Override
-    public DataType dataType() {
-        return DataTypes.DOUBLE;
-    }
-
-    @Override
-    public String innerName() {
-        Double value = (Double) SqlDataTypeConverter.convert(Foldables.valueOf(percent), DataTypes.DOUBLE);
-        return Double.toString(value);
     }
 }

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/aggregate/PercentileAggregate.java

@@ -0,0 +1,202 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.sql.expression.function.aggregate;
+
+import org.elasticsearch.search.aggregations.metrics.PercentilesConfig;
+import org.elasticsearch.search.aggregations.metrics.PercentilesMethod;
+import org.elasticsearch.xpack.ql.expression.Expression;
+import org.elasticsearch.xpack.ql.expression.Expressions;
+import org.elasticsearch.xpack.ql.expression.Expressions.ParamOrdinal;
+import org.elasticsearch.xpack.ql.expression.Foldables;
+import org.elasticsearch.xpack.ql.expression.TypeResolutions;
+import org.elasticsearch.xpack.ql.expression.function.TwoOptionalArguments;
+import org.elasticsearch.xpack.ql.expression.function.aggregate.EnclosedAgg;
+import org.elasticsearch.xpack.ql.tree.Source;
+import org.elasticsearch.xpack.ql.type.DataType;
+import org.elasticsearch.xpack.ql.type.DataTypes;
+import org.elasticsearch.xpack.sql.type.SqlDataTypeConverter;
+
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.Function;
+
+import static java.util.Collections.singletonList;
+import static org.elasticsearch.common.logging.LoggerMessageFormat.format;
+import static org.elasticsearch.xpack.ql.expression.TypeResolutions.isFoldable;
+import static org.elasticsearch.xpack.ql.expression.TypeResolutions.isNumeric;
+
+abstract class PercentileAggregate extends NumericAggregate implements EnclosedAgg, TwoOptionalArguments {
+
+    private static final PercentilesConfig.TDigest DEFAULT_PERCENTILES_CONFIG = new PercentilesConfig.TDigest();
+
+    // preferred method name to configurator mapping (type resolution, method parameter -> config)
+    // contains all the possible PercentilesMethods that we know of and are capable of parameterizing at the moment
+    private static final Map<String, MethodConfigurator> METHOD_CONFIGURATORS = new LinkedHashMap<>();
+    static {
+        Arrays.asList(
+            new MethodConfigurator(PercentilesMethod.TDIGEST, TypeResolutions::isNumeric, methodParameter -> {
+                Double compression = foldNullSafe(methodParameter, DataTypes.DOUBLE);
+                return compression == null ? new PercentilesConfig.TDigest() : new PercentilesConfig.TDigest(compression);
+            }), 
+            new MethodConfigurator(PercentilesMethod.HDR, TypeResolutions::isInteger, methodParameter -> {
+                Integer numOfDigits = foldNullSafe(methodParameter, DataTypes.INTEGER);
+                return numOfDigits == null ? new PercentilesConfig.Hdr() : new PercentilesConfig.Hdr(numOfDigits);
+            }))
+            .forEach(c -> METHOD_CONFIGURATORS.put(c.method.getParseField().getPreferredName(), c));
+    }
+
+    private static class MethodConfigurator {
+
+        @FunctionalInterface
+        private interface MethodParameterResolver {
+            TypeResolution resolve(Expression methodParameter, String sourceText, ParamOrdinal methodParameterOrdinal);
+        }
+
+        private final PercentilesMethod method;
+        private final MethodParameterResolver resolver;
+        private final Function<Expression, PercentilesConfig> parameterToConfig;
+
+        MethodConfigurator(
+            PercentilesMethod method, MethodParameterResolver resolver, Function<Expression, PercentilesConfig> parameterToConfig) {
+            this.method = method;
+            this.resolver = resolver;
+            this.parameterToConfig = parameterToConfig;
+        }
+
+    }
+    
+    private final Expression parameter;
+    private final Expression method;
+    private final Expression methodParameter;
+
+    PercentileAggregate(Source source, Expression field, Expression parameter, Expression method, Expression methodParameter) 
+    {
+        super(source, field, singletonList(parameter));
+        this.parameter = parameter;
+        this.method = method;
+        this.methodParameter = methodParameter;
+    }
+
+    @Override 
+    protected TypeResolution resolveType() {        
+        TypeResolution resolution = super.resolveType();
+        if (resolution.unresolved()) {
+            return resolution;
+        }
+
+        resolution = isFoldable(parameter, sourceText(), ParamOrdinal.SECOND);
+        if (resolution.unresolved()) {
+            return resolution;
+        }
+
+        resolution = isNumeric(parameter, sourceText(), ParamOrdinal.SECOND);
+        if (resolution.unresolved()) {
+            return resolution;
+        }
+
+        ParamOrdinal methodOrdinal = ParamOrdinal.fromIndex(parameters().size() + 1);
+        ParamOrdinal methodParameterOrdinal = ParamOrdinal.fromIndex(parameters().size() + 2);
+        
+        if (method != null) {
+            resolution = isFoldable(method, sourceText(), methodOrdinal);
+            if (resolution.unresolved()) {
+                return resolution;
+            }
+            resolution = TypeResolutions.isString(method, sourceText(), methodOrdinal);
+            if (resolution.unresolved()) {
+                return resolution;
+            }
+
+            String methodName = (String) method.fold();
+
+            MethodConfigurator methodConfigurator = METHOD_CONFIGURATORS.get(methodName);
+            if (methodConfigurator == null) {
+                return new TypeResolution(format(null, "{}argument of [{}] must be one of {}, received [{}]",
+                    methodOrdinal.name().toLowerCase(Locale.ROOT) + " ", sourceText(),
+                    METHOD_CONFIGURATORS.keySet(), methodName));
+            }
+
+            // if method is null, the method parameter is not checked
+            if (methodParameter != null && Expressions.isNull(methodParameter) == false) {
+                resolution = isFoldable(methodParameter, sourceText(), methodParameterOrdinal);
+                if (resolution.unresolved()) {
+                    return resolution;
+                }
+
+                resolution = methodConfigurator.resolver.resolve(methodParameter, sourceText(), methodParameterOrdinal);
+                return resolution;
+            }
+        }
+
+        return TypeResolution.TYPE_RESOLVED;
+    }
+    
+    public Expression parameter() {
+        return parameter;
+    }
+
+    public Expression method() {
+        return method;
+    }
+
+    public Expression methodParameter() {
+        return methodParameter;
+    }
+
+    @Override
+    public DataType dataType() {
+        return DataTypes.DOUBLE;
+    }
+
+    @Override
+    public String innerName() {
+        Double value = (Double) SqlDataTypeConverter.convert(Foldables.valueOf(parameter), DataTypes.DOUBLE);
+        return Double.toString(value);
+    }
+
+    public PercentilesConfig percentilesConfig() {
+        if (method == null) {
+            // sadly we had to set the default here, the PercentilesConfig does not provide a default
+            return DEFAULT_PERCENTILES_CONFIG;
+        }
+        String methodName = foldNullSafe(method, DataTypes.KEYWORD);
+        MethodConfigurator methodConfigurator = METHOD_CONFIGURATORS.get(methodName);
+        if (methodConfigurator == null) {
+            throw new IllegalStateException("Not handled PercentilesMethod [" + methodName + "], type resolution needs fix");
+        }
+        return methodConfigurator.parameterToConfig.apply(methodParameter);
+    }
+
+    @SuppressWarnings("unchecked")
+    private static <T> T foldNullSafe(Expression e, DataType dataType) {
+        return e == null ? null : (T) SqlDataTypeConverter.convert(Foldables.valueOf(e), dataType);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(super.hashCode(), method, methodParameter);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        
+        if (!super.equals(o)) {
+            return false;
+        }
+
+        PercentileAggregate that = (PercentileAggregate) o;
+
+        return Objects.equals(method, that.method)
+            && Objects.equals(methodParameter, that.methodParameter);
+    }
+}

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/aggregate/PercentileCompoundAggregate.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.sql.expression.function.aggregate;
+
+import org.elasticsearch.search.aggregations.metrics.PercentilesConfig;
+import org.elasticsearch.xpack.ql.expression.Expression;
+import org.elasticsearch.xpack.ql.tree.Source;
+
+import java.util.List;
+import java.util.Objects;
+
+public abstract class PercentileCompoundAggregate extends CompoundNumericAggregate {
+    protected final PercentilesConfig percentilesConfig;
+
+    public PercentileCompoundAggregate(
+        Source source, Expression field, List<Expression> arguments, PercentilesConfig percentilesConfig) {
+        super(source, field, arguments);
+        this.percentilesConfig = percentilesConfig;
+    }
+
+    public PercentilesConfig percentilesConfig() {
+        return percentilesConfig;
+    }
+
+    @Override 
+    public int hashCode() {
+        return Objects.hash(super.hashCode(), percentilesConfig);
+    }
+
+    @Override 
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (!super.equals(o)) {
+            return false;
+        }
+
+        return Objects.equals(percentilesConfig, ((Percentiles) o).percentilesConfig);
+    }
+}

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/aggregate/PercentileRank.java

@@ -6,34 +6,20 @@
 package org.elasticsearch.xpack.sql.expression.function.aggregate;
 
 import org.elasticsearch.xpack.ql.expression.Expression;
-import org.elasticsearch.xpack.ql.expression.Expressions.ParamOrdinal;
-import org.elasticsearch.xpack.ql.expression.Foldables;
-import org.elasticsearch.xpack.ql.expression.function.aggregate.AggregateFunction;
-import org.elasticsearch.xpack.ql.expression.function.aggregate.EnclosedAgg;
 import org.elasticsearch.xpack.ql.tree.NodeInfo;
 import org.elasticsearch.xpack.ql.tree.Source;
-import org.elasticsearch.xpack.ql.type.DataType;
-import org.elasticsearch.xpack.ql.type.DataTypes;
-import org.elasticsearch.xpack.sql.type.SqlDataTypeConverter;
 
 import java.util.List;
 
-import static java.util.Collections.singletonList;
+public class PercentileRank extends PercentileAggregate {
-import static org.elasticsearch.xpack.ql.expression.TypeResolutions.isFoldable;
-import static org.elasticsearch.xpack.ql.expression.TypeResolutions.isNumeric;
 
-public class PercentileRank extends AggregateFunction implements EnclosedAgg {
+    public PercentileRank(Source source, Expression field, Expression value, Expression method, Expression methodParameter) {
-
+        super(source, field, value, method, methodParameter);
-    private final Expression value;
-
-    public PercentileRank(Source source, Expression field, Expression value) {
-        super(source, field, singletonList(value));
-        this.value = value;
     }
 
     @Override
     protected NodeInfo<PercentileRank> info() {
-        return NodeInfo.create(this, PercentileRank::new, field(), value);
+        return NodeInfo.create(this, PercentileRank::new, field(), value(), method(), methodParameter());
     }
 
     @Override
@@ -41,36 +27,10 @@ public class PercentileRank extends AggregateFunction implements EnclosedAgg {
         if (newChildren.size() != 2) {
             throw new IllegalArgumentException("expected [2] children but received [" + newChildren.size() + "]");
         }
-        return new PercentileRank(source(), newChildren.get(0), newChildren.get(1));
+        return new PercentileRank(source(), newChildren.get(0), newChildren.get(1), method(), methodParameter());
-    }
-
-    @Override
-    protected TypeResolution resolveType() {
-        TypeResolution resolution = isFoldable(value, sourceText(), ParamOrdinal.SECOND);
-        if (resolution.unresolved()) {
-            return resolution;
-        }
-
-        resolution = super.resolveType();
-        if (resolution.unresolved()) {
-            return resolution;
-        }
-
-        return isNumeric(value, sourceText(), ParamOrdinal.DEFAULT);
     }
 
     public Expression value() {
-        return value;
+        return parameter();
-    }
-
-    @Override
-    public DataType dataType() {
-        return DataTypes.DOUBLE;
-    }
-
-    @Override
-    public String innerName() {
-        Double doubleValue = (Double) SqlDataTypeConverter.convert(Foldables.valueOf(value), DataTypes.DOUBLE);
-        return Double.toString(doubleValue);
     }
 }

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/aggregate/PercentileRanks.java

@@ -5,24 +5,22 @@
  */
 package org.elasticsearch.xpack.sql.expression.function.aggregate;
 
+import org.elasticsearch.search.aggregations.metrics.PercentilesConfig;
 import org.elasticsearch.xpack.ql.expression.Expression;
 import org.elasticsearch.xpack.ql.tree.NodeInfo;
 import org.elasticsearch.xpack.ql.tree.Source;
 
 import java.util.List;
 
-public class PercentileRanks extends CompoundNumericAggregate {
+public class PercentileRanks extends PercentileCompoundAggregate {
 
-    private final List<Expression> values;
+    public PercentileRanks(Source source, Expression field, List<Expression> values, PercentilesConfig percentilesConfig) {
-
+        super(source, field, values, percentilesConfig);
-    public PercentileRanks(Source source, Expression field, List<Expression> values) {
-        super(source, field, values);
-        this.values = values;
     }
 
     @Override
     protected NodeInfo<PercentileRanks> info() {
-        return NodeInfo.create(this, PercentileRanks::new, field(), values);
+        return NodeInfo.create(this, PercentileRanks::new, field(), values(), percentilesConfig);
     }
 
     @Override
@@ -30,10 +28,12 @@ public class PercentileRanks extends CompoundNumericAggregate {
         if (newChildren.size() < 2) {
             throw new IllegalArgumentException("expected at least [2] children but received [" + newChildren.size() + "]");
         }
-        return new PercentileRanks(source(), newChildren.get(0), newChildren.subList(1, newChildren.size()));
+        return new PercentileRanks(source(), newChildren.get(0), newChildren.subList(1, newChildren.size()), percentilesConfig);
+    }
+    
+    @SuppressWarnings("unchecked")
+    public List<Expression> values() {
+        return (List<Expression>) parameters();
     }
 
-    public List<Expression> values() {
-        return values;
-    }
 }

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/aggregate/Percentiles.java

@@ -5,35 +5,35 @@
  */
 package org.elasticsearch.xpack.sql.expression.function.aggregate;
 
+import org.elasticsearch.search.aggregations.metrics.PercentilesConfig;
 import org.elasticsearch.xpack.ql.expression.Expression;
 import org.elasticsearch.xpack.ql.tree.NodeInfo;
 import org.elasticsearch.xpack.ql.tree.Source;
 
 import java.util.List;
 
-public class Percentiles extends CompoundNumericAggregate {
+public class Percentiles extends PercentileCompoundAggregate {
 
-    private final List<Expression> percents;
+    public Percentiles(Source source, Expression field, List<Expression> percents, PercentilesConfig percentilesConfig) {
-
+        super(source, field, percents, percentilesConfig);
-    public Percentiles(Source source, Expression field, List<Expression> percents) {
-        super(source, field, percents);
-        this.percents = percents;
     }
 
     @Override
     protected NodeInfo<Percentiles> info() {
-        return NodeInfo.create(this, Percentiles::new, field(), percents);
+        return NodeInfo.create(this, Percentiles::new, field(), percents(), percentilesConfig());
     }
 
     @Override
     public Percentiles replaceChildren(List<Expression> newChildren) {
         if (newChildren.size() < 2) {
-            throw new IllegalArgumentException("expected more than one child but received [" + newChildren.size() + "]");
+            throw new IllegalArgumentException("expected at least [2] children but received [" + newChildren.size() + "]");
         }
-        return new Percentiles(source(), newChildren.get(0), newChildren.subList(1, newChildren.size()));
+        return new Percentiles(source(), newChildren.get(0), newChildren.subList(1, newChildren.size()), percentilesConfig());
     }
 
+    @SuppressWarnings("unchecked") 
     public List<Expression> percents() {
-        return percents;
+        return (List<Expression>) parameters();
     }
+
 }

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/optimizer/Optimizer.java

@@ -5,6 +5,8 @@
  */
 package org.elasticsearch.xpack.sql.optimizer;
 
+import org.elasticsearch.common.collect.Tuple;
+import org.elasticsearch.search.aggregations.metrics.PercentilesConfig;
 import org.elasticsearch.xpack.ql.expression.Alias;
 import org.elasticsearch.xpack.ql.expression.Attribute;
 import org.elasticsearch.xpack.ql.expression.AttributeMap;
@@ -1016,39 +1018,50 @@ public class Optimizer extends RuleExecutor<LogicalPlan> {
         }
     }
 
+    private static class PercentileKey extends Tuple<Expression, PercentilesConfig> {
+        PercentileKey(Percentile per) {
+            super(per.field(), per.percentilesConfig());
+        }
+
+        PercentileKey(PercentileRank per) {
+            super(per.field(), per.percentilesConfig());
+        }
+        
+        private Expression field() {
+            return v1();
+        }
+        
+        private PercentilesConfig percentilesConfig() {
+            return v2();
+        }
+    }
+
     static class ReplaceAggsWithPercentiles extends OptimizerBasicRule {
 
         @Override
         public LogicalPlan apply(LogicalPlan p) {
-            // percentile per field/expression
+            Map<PercentileKey, Set<Expression>> percentsPerAggKey = new LinkedHashMap<>();
-            Map<Expression, Set<Expression>> percentsPerField = new LinkedHashMap<>();
 
-            // count gather the percents for each field
             p.forEachExpressionsUp(e -> {
                 if (e instanceof Percentile) {
                     Percentile per = (Percentile) e;
-                    Expression field = per.field();
+                    percentsPerAggKey.computeIfAbsent(new PercentileKey(per), v -> new LinkedHashSet<>())
-                    Set<Expression> percentiles = percentsPerField.get(field);
+                        .add(per.percent());
-
-                    if (percentiles == null) {
-                        percentiles = new LinkedHashSet<>();
-                        percentsPerField.put(field, percentiles);
-                    }
-
-                    percentiles.add(per.percent());
                 }
             });
 
-            Map<Expression, Percentiles> percentilesPerField = new LinkedHashMap<>();
+            // create a Percentile agg for each agg key
-            // create a Percentile agg for each field (and its associated percents)
+            Map<PercentileKey, Percentiles> percentilesPerAggKey = new LinkedHashMap<>();
-            percentsPerField.forEach((k, v) -> {
+            percentsPerAggKey.forEach((aggKey, percents) -> percentilesPerAggKey.put(
-                percentilesPerField.put(k, new Percentiles(v.iterator().next().source(), k, new ArrayList<>(v)));
+                    aggKey,
-            });
+                    new Percentiles(percents.iterator().next().source(), aggKey.field(), new ArrayList<>(percents), 
+                        aggKey.percentilesConfig())));
 
             return p.transformExpressionsUp(e -> {
                 if (e instanceof Percentile) {
                     Percentile per = (Percentile) e;
-                    Percentiles percentiles = percentilesPerField.get(per.field());
+                    PercentileKey a = new PercentileKey(per);
+                    Percentiles percentiles = percentilesPerAggKey.get(a);
                     return new InnerAggregate(per, percentiles);
                 }
 
@@ -1061,35 +1074,27 @@ public class Optimizer extends RuleExecutor<LogicalPlan> {
 
         @Override
         public LogicalPlan apply(LogicalPlan p) {
-            // percentile per field/expression
+            final Map<PercentileKey, Set<Expression>> valuesPerAggKey = new LinkedHashMap<>();
-            final Map<Expression, Set<Expression>> percentPerField = new LinkedHashMap<>();
 
-            // count gather the percents for each field
             p.forEachExpressionsUp(e -> {
                 if (e instanceof PercentileRank) {
                     PercentileRank per = (PercentileRank) e;
-                    Expression field = per.field();
+                    valuesPerAggKey.computeIfAbsent(new PercentileKey(per), v -> new LinkedHashSet<>())
-                    Set<Expression> percentiles = percentPerField.get(field);
+                        .add(per.value());
-
-                    if (percentiles == null) {
-                        percentiles = new LinkedHashSet<>();
-                        percentPerField.put(field, percentiles);
-                    }
-
-                    percentiles.add(per.value());
                 }
             });
 
-            Map<Expression, PercentileRanks> ranksPerField = new LinkedHashMap<>();
+            // create a PercentileRank agg for each agg key
-            // create a PercentileRanks agg for each field (and its associated values)
+            Map<PercentileKey, PercentileRanks> ranksPerAggKey = new LinkedHashMap<>();
-            percentPerField.forEach((k, v) -> {
+            valuesPerAggKey.forEach((aggKey, values) -> ranksPerAggKey.put(
-                ranksPerField.put(k, new PercentileRanks(v.iterator().next().source(), k, new ArrayList<>(v)));
+                aggKey,
-            });
+                new PercentileRanks(values.iterator().next().source(), aggKey.field(), new ArrayList<>(values), 
+                    aggKey.percentilesConfig())));
 
             return p.transformExpressionsUp(e -> {
                 if (e instanceof PercentileRank) {
                     PercentileRank per = (PercentileRank) e;
-                    PercentileRanks ranks = ranksPerField.get(per.field());
+                    PercentileRanks ranks = ranksPerAggKey.get(new PercentileKey(per));
                     return new InnerAggregate(per, ranks);
                 }
 

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/planner/QueryTranslator.java

@@ -612,7 +612,7 @@ final class QueryTranslator {
 
         @Override
         protected LeafAgg toAgg(String id, Percentiles p) {
-            return new PercentilesAgg(id, asFieldOrLiteralOrScript(p), foldAndConvertToDoubles(p.percents()));
+            return new PercentilesAgg(id, asFieldOrLiteralOrScript(p), foldAndConvertToDoubles(p.percents()), p.percentilesConfig());
         }
     }
 
@@ -620,7 +620,7 @@ final class QueryTranslator {
 
         @Override
         protected LeafAgg toAgg(String id, PercentileRanks p) {
-            return new PercentileRanksAgg(id, asFieldOrLiteralOrScript(p), foldAndConvertToDoubles(p.values()));
+            return new PercentileRanksAgg(id, asFieldOrLiteralOrScript(p), foldAndConvertToDoubles(p.values()), p.percentilesConfig());
         }
     }
 
@@ -681,4 +681,5 @@ final class QueryTranslator {
         }
         return values;
     }
+
 }

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/querydsl/agg/PercentileRanksAgg.java

@@ -5,6 +5,7 @@
  */
 package org.elasticsearch.xpack.sql.querydsl.agg;
 
+import org.elasticsearch.search.aggregations.metrics.PercentilesConfig;
 import org.elasticsearch.search.aggregations.support.ValuesSourceAggregationBuilder;
 
 import java.util.List;
@@ -15,14 +16,17 @@ import static org.elasticsearch.search.aggregations.AggregationBuilders.percenti
 public class PercentileRanksAgg extends DefaultAggSourceLeafAgg {
 
     private final List<Double> values;
+    private final PercentilesConfig percentilesConfig;
 
-    public PercentileRanksAgg(String id, AggSource source, List<Double> values) {
+    public PercentileRanksAgg(String id, AggSource source, List<Double> values, PercentilesConfig percentilesConfig) {
         super(id, source);
         this.values = values;
+        this.percentilesConfig = percentilesConfig;
     }
 
     @Override
     Function<String, ValuesSourceAggregationBuilder<?>> builder() {
-        return s -> percentileRanks(s, values.stream().mapToDouble(Double::doubleValue).toArray());
+        return s -> percentileRanks(s, values.stream().mapToDouble(Double::doubleValue).toArray())
+            .percentilesConfig(percentilesConfig);
     }
 }

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/querydsl/agg/PercentilesAgg.java

@@ -5,32 +5,29 @@
  */
 package org.elasticsearch.xpack.sql.querydsl.agg;
 
-import org.elasticsearch.search.aggregations.AggregationBuilder;
+import org.elasticsearch.search.aggregations.metrics.PercentilesConfig;
-import org.elasticsearch.search.aggregations.AggregationBuilders;
-import org.elasticsearch.search.aggregations.metrics.PercentilesAggregationBuilder;
 import org.elasticsearch.search.aggregations.support.ValuesSourceAggregationBuilder;
 
 import java.util.List;
 import java.util.function.Function;
 
+import static org.elasticsearch.search.aggregations.AggregationBuilders.percentiles;
+
 public class PercentilesAgg extends DefaultAggSourceLeafAgg {
 
     private final List<Double> percents;
+    private final PercentilesConfig percentilesConfig;
 
-    public PercentilesAgg(String id, AggSource source, List<Double> percents) {
+    public PercentilesAgg(String id, AggSource source, List<Double> percents, PercentilesConfig percentilesConfig) {
         super(id, source);
         this.percents = percents;
-    }
+        this.percentilesConfig = percentilesConfig;
-    
-    @Override
-    AggregationBuilder toBuilder() {
-        // TODO: look at keyed
-        PercentilesAggregationBuilder builder = (PercentilesAggregationBuilder) super.toBuilder();
-        return builder.percentiles(percents.stream().mapToDouble(Double::doubleValue).toArray());
     }
 
     @Override
     Function<String, ValuesSourceAggregationBuilder<?>> builder() {
-        return AggregationBuilders::percentiles;
+        return s -> percentiles(s)
+            .percentiles(percents.stream().mapToDouble(Double::doubleValue).toArray())
+            .percentilesConfig(percentilesConfig);
     }
 }

x-pack/plugin/sql/src/test/java/org/elasticsearch/xpack/sql/analysis/analyzer/VerifierErrorMessagesTests.java

@@ -974,11 +974,64 @@ public class VerifierErrorMessagesTests extends ESTestCase {
             error("SELECT PERCENTILE(int, ABS(int)) FROM test"));
     }
 
+    public void testErrorMessageForPercentileWithWrongMethodType() {
+        assertEquals("1:8: third argument of [PERCENTILE(int, 50, 2)] must be [string], found value [2] type [integer]",
+            error("SELECT PERCENTILE(int, 50, 2) FROM test"));
+    }
+
+    public void testErrorMessageForPercentileWithNullMethodType() {
+        assertEquals("1:8: third argument of [PERCENTILE(int, 50, null)] must be one of [tdigest, hdr], received [null]",
+            error("SELECT PERCENTILE(int, 50, null) FROM test"));
+    }
+
+    public void testErrorMessageForPercentileWithHDRRequiresInt() {
+        assertEquals("1:8: fourth argument of [PERCENTILE(int, 50, 'hdr', 2.2)] must be [integer], found value [2.2] type [double]",
+            error("SELECT PERCENTILE(int, 50, 'hdr', 2.2) FROM test"));
+    }
+
+    public void testErrorMessageForPercentileWithWrongMethod() {
+        assertEquals("1:8: third argument of [PERCENTILE(int, 50, 'notExistingMethod', 5)] must be " +
+                "one of [tdigest, hdr], received [notExistingMethod]",
+            error("SELECT PERCENTILE(int, 50, 'notExistingMethod', 5) FROM test"));
+    }
+
+    public void testErrorMessageForPercentileWithWrongMethodParameterType() {
+        assertEquals("1:8: fourth argument of [PERCENTILE(int, 50, 'tdigest', '5')] must be [numeric], found value ['5'] type [keyword]",
+            error("SELECT PERCENTILE(int, 50, 'tdigest', '5') FROM test"));
+    }
+
     public void testErrorMessageForPercentileRankWithSecondArgBasedOnAField() {
         assertEquals("1:8: second argument of [PERCENTILE_RANK(int, ABS(int))] must be a constant, received [ABS(int)]",
             error("SELECT PERCENTILE_RANK(int, ABS(int)) FROM test"));
     }
 
+    public void testErrorMessageForPercentileRankWithWrongMethodType() {
+        assertEquals("1:8: third argument of [PERCENTILE_RANK(int, 50, 2)] must be [string], found value [2] type [integer]",
+            error("SELECT PERCENTILE_RANK(int, 50, 2) FROM test"));
+    }
+
+    public void testErrorMessageForPercentileRankWithNullMethodType() {
+        assertEquals("1:8: third argument of [PERCENTILE_RANK(int, 50, null)] must be one of [tdigest, hdr], received [null]",
+            error("SELECT PERCENTILE_RANK(int, 50, null) FROM test"));
+    }
+
+    public void testErrorMessageForPercentileRankWithHDRRequiresInt() {
+        assertEquals("1:8: fourth argument of [PERCENTILE_RANK(int, 50, 'hdr', 2.2)] must be [integer], found value [2.2] type [double]",
+            error("SELECT PERCENTILE_RANK(int, 50, 'hdr', 2.2) FROM test"));
+    }
+
+    public void testErrorMessageForPercentileRankWithWrongMethod() {
+        assertEquals("1:8: third argument of [PERCENTILE_RANK(int, 50, 'notExistingMethod', 5)] must be " +
+                "one of [tdigest, hdr], received [notExistingMethod]",
+            error("SELECT PERCENTILE_RANK(int, 50, 'notExistingMethod', 5) FROM test"));
+    }
+
+    public void testErrorMessageForPercentileRankWithWrongMethodParameterType() {
+        assertEquals("1:8: fourth argument of [PERCENTILE_RANK(int, 50, 'tdigest', '5')] must be [numeric], " +
+                "found value ['5'] type [keyword]",
+            error("SELECT PERCENTILE_RANK(int, 50, 'tdigest', '5') FROM test"));
+    }
+
     public void testTopHitsFirstArgConstant() {
         assertEquals("1:8: first argument of [FIRST('foo', int)] must be a table column, found constant ['foo']",
             error("SELECT FIRST('foo', int) FROM test"));

x-pack/plugin/sql/src/test/java/org/elasticsearch/xpack/sql/planner/QueryTranslatorTests.java

@@ -10,8 +10,12 @@ import org.elasticsearch.common.time.DateFormatter;
 import org.elasticsearch.index.query.ExistsQueryBuilder;
 import org.elasticsearch.search.aggregations.AggregationBuilder;
 import org.elasticsearch.search.aggregations.bucket.filter.FilterAggregationBuilder;
+import org.elasticsearch.search.aggregations.metrics.AbstractPercentilesAggregationBuilder;
 import org.elasticsearch.search.aggregations.metrics.AvgAggregationBuilder;
 import org.elasticsearch.search.aggregations.metrics.CardinalityAggregationBuilder;
+import org.elasticsearch.search.aggregations.metrics.PercentileRanksAggregationBuilder;
+import org.elasticsearch.search.aggregations.metrics.PercentilesAggregationBuilder;
+import org.elasticsearch.search.aggregations.metrics.PercentilesConfig;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xpack.ql.QlIllegalArgumentException;
 import org.elasticsearch.xpack.ql.execution.search.FieldExtraction;
@@ -86,6 +90,10 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.function.BiConsumer;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 import java.util.stream.Stream;
 
 import static org.elasticsearch.xpack.ql.type.DataTypes.BOOLEAN;
@@ -104,6 +112,7 @@ import static org.elasticsearch.xpack.sql.util.DateUtils.UTC;
 import static org.hamcrest.CoreMatchers.containsString;
 import static org.hamcrest.Matchers.endsWith;
 import static org.hamcrest.Matchers.everyItem;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.startsWith;
 
@@ -2308,4 +2317,92 @@ public class QueryTranslatorTests extends ESTestCase {
         assertEquals(1, eqe.output().size());
         assertThat(eqe.queryContainer().toString().replaceAll("\\s+", ""), containsString("\"terms\":{\"int\":[1,2,3],"));
     }
+
+    @SuppressWarnings({"rawtypes"})
+    private static List<AbstractPercentilesAggregationBuilder> percentilesAggsByField(PhysicalPlan p, int fieldCount) {
+        assertEquals(EsQueryExec.class, p.getClass());
+        EsQueryExec ee = (EsQueryExec) p;
+        AggregationBuilder aggregationBuilder = ee.queryContainer().aggs().asAggBuilder();
+        assertEquals(fieldCount, ee.output().size());
+        assertEquals(ReferenceAttribute.class, ee.output().get(0).getClass());
+        assertEquals(fieldCount, ee.queryContainer().fields().size());
+        assertThat(fieldCount, greaterThanOrEqualTo(ee.queryContainer().aggs().asAggBuilder().getSubAggregations().size()));
+        Map<String, AggregationBuilder> aggsByName =
+            aggregationBuilder.getSubAggregations().stream().collect(Collectors.toMap(AggregationBuilder::getName, ab -> ab));
+        return IntStream.range(0, fieldCount).mapToObj(i -> {
+            String percentileAggName = ((MetricAggRef) ee.queryContainer().fields().get(i).v1()).name();
+            return (AbstractPercentilesAggregationBuilder) aggsByName.get(percentileAggName);
+        }).collect(Collectors.toList());
+    }
+
+    @SuppressWarnings({"rawtypes"})
+    public void testPercentileMethodParametersSameAsDefault() {
+        BiConsumer<String, Function<AbstractPercentilesAggregationBuilder, double[]>> test = (fnName, pctOrValFn) -> {
+            final int fieldCount = 5;
+            final String sql = ("SELECT " +
+                // 0-3: these all should fold into the same aggregation
+                "   PERCENTILE(int, 50, 'tdigest', 79.8 + 20.2), " +
+                "   PERCENTILE(int, 40 + 10, 'tdigest', null), " +
+                "   PERCENTILE(int, 50, 'tdigest'), " +
+                "   PERCENTILE(int, 50), " +
+                // 4: this has a different method parameter
+                // just to make sure we don't fold everything to default
+                "   PERCENTILE(int, 50, 'tdigest', 22) " 
+                + "FROM test").replaceAll("PERCENTILE", fnName);
+            
+            List<AbstractPercentilesAggregationBuilder> aggs = percentilesAggsByField(optimizeAndPlan(sql), fieldCount);
+            
+            // 0-3
+            assertEquals(aggs.get(0), aggs.get(1));
+            assertEquals(aggs.get(0), aggs.get(2));
+            assertEquals(aggs.get(0), aggs.get(3));
+            assertEquals(new PercentilesConfig.TDigest(), aggs.get(0).percentilesConfig());
+            assertArrayEquals(new double[] { 50 }, pctOrValFn.apply(aggs.get(0)), 0);
+
+            // 4
+            assertEquals(new PercentilesConfig.TDigest(22), aggs.get(4).percentilesConfig());
+            assertArrayEquals(new double[] { 50 }, pctOrValFn.apply(aggs.get(4)), 0);
+        };
+        
+        test.accept("PERCENTILE", p -> ((PercentilesAggregationBuilder)p).percentiles());
+        test.accept("PERCENTILE_RANK", p -> ((PercentileRanksAggregationBuilder)p).values());
+    }
+
+    @SuppressWarnings({"rawtypes"})
+    public void testPercentileOptimization() {
+        BiConsumer<String, Function<AbstractPercentilesAggregationBuilder, double[]>> test = (fnName, pctOrValFn) -> {
+            final int fieldCount = 5;
+            final String sql = ("SELECT " +
+                // 0-1: fold into the same aggregation
+                "   PERCENTILE(int, 50, 'tdigest'), " +
+                "   PERCENTILE(int, 60, 'tdigest'), " +
+                
+                // 2-3: fold into one aggregation
+                "   PERCENTILE(int, 50, 'hdr'), " +
+                "   PERCENTILE(int, 60, 'hdr', 3), " +
+                
+                // 4: folds into a separate aggregation
+                "   PERCENTILE(int, 60, 'hdr', 4)" +
+                "FROM test").replaceAll("PERCENTILE", fnName);
+
+            List<AbstractPercentilesAggregationBuilder> aggs = percentilesAggsByField(optimizeAndPlan(sql), fieldCount);
+            
+            // 0-1
+            assertEquals(aggs.get(0), aggs.get(1));
+            assertEquals(new PercentilesConfig.TDigest(), aggs.get(0).percentilesConfig());
+            assertArrayEquals(new double[]{50, 60}, pctOrValFn.apply(aggs.get(0)), 0);
+
+            // 2-3
+            assertEquals(aggs.get(2), aggs.get(3));
+            assertEquals(new PercentilesConfig.Hdr(), aggs.get(2).percentilesConfig());
+            assertArrayEquals(new double[]{50, 60}, pctOrValFn.apply(aggs.get(2)), 0);
+            
+            // 4
+            assertEquals(new PercentilesConfig.Hdr(4), aggs.get(4).percentilesConfig());
+            assertArrayEquals(new double[]{60}, pctOrValFn.apply(aggs.get(4)), 0);
+        };
+
+        test.accept("PERCENTILE", p -> ((PercentilesAggregationBuilder)p).percentiles());
+        test.accept("PERCENTILE_RANK", p -> ((PercentileRanksAggregationBuilder)p).values());
+    }
 }