Aggregations: Added percentile rank aggregation

Percentile Rank Aggregation is the reverse of the Percetiles aggregation.  It determines the percentile rank (the proportion of values less than a given value) of the provided array of values.

Closes #6386

docs/reference/search/aggregations/metrics.asciidoc

@@ -16,6 +16,8 @@ include::metrics/valuecount-aggregation.asciidoc[]
 
 include::metrics/percentile-aggregation.asciidoc[]
 
+include::metrics/percentile-rank-aggregation.asciidoc[]
+
 include::metrics/cardinality-aggregation.asciidoc[]
 
 include::metrics/geobounds-aggregation.asciidoc[]

docs/reference/search/aggregations/metrics/percentile-aggregation.asciidoc

@@ -125,6 +125,7 @@ a script to convert them on-the-fly:
 script to generate values which percentiles are calculated on
 <2> Scripting supports parameterized input just like any other script
 
+[[search-aggregations-metrics-percentile-aggregation-approximation]]
 ==== Percentiles are (usually) approximate
 
 There are many different algorithms to calculate percentiles.  The naive
@@ -161,6 +162,7 @@ for large number of values is that the law of large numbers makes the distributi
 values more and more uniform and the t-digest tree can do a better job at summarizing
 it. It would not be the case on more skewed distributions.
 
+[[search-aggregations-metrics-percentile-aggregation-compression]]
 ==== Compression
 
 Approximate algorithms must balance memory utilization with estimation accuracy.

docs/reference/search/aggregations/metrics/percentile-rank-aggregation.asciidoc

@@ -0,0 +1,95 @@
+[[search-aggregations-metrics-percentile-rank-aggregation]]
+=== Percentile Ranks Aggregation
+
+coming[1.3.0]
+
+A `multi-value` metrics aggregation that calculates one or more percentile ranks
+over numeric values extracted from the aggregated documents.  These values
+can be extracted either from specific numeric fields in the documents, or
+be generated by a provided script.
+
+.Experimental!
+[IMPORTANT]
+=====
+This feature is marked as experimental, and may be subject to change in the
+future.  If you use this feature, please let us know your experience with it!
+=====
+
+[NOTE]
+==================================================
+Please see <<search-aggregations-metrics-percentile-aggregation-approximation>> 
+and <<search-aggregations-metrics-percentile-aggregation-compression>> for advice 
+regarding approximation and memory use of the percentile ranks aggregation
+==================================================
+
+Percentile rank show the percentage of observed values which are below certain 
+value.  For example, if a value is greater than or equal to 95% of the observed values
+it is said to be at the 95th percentile rank.
+
+Assume your data consists of website load times.  You may have a service agreement that 
+95% of page loads completely within 15ms and 99% of page loads complete within 30ms.
+
+Let's look at a range of percentiles representing load time:
+
+[source,js]
+--------------------------------------------------
+{
+    "aggs" : {
+        "load_time_outlier" : {
+            "percentile_ranks" : {
+                "field" : "load_time" <1>
+                "values" : [15, 30]
+            }
+        }
+    }
+}
+--------------------------------------------------
+<1> The field `load_time` must be a numeric field
+
+The response will look like this:
+
+[source,js]
+--------------------------------------------------
+{
+    ...
+
+   "aggregations": {
+      "load_time_outlier": {
+         "values" : {
+            "15": 92,
+            "30": 100
+         }
+      }
+   }
+}
+--------------------------------------------------
+
+From this information you can determine you are hitting the 99% load time target but not quite 
+hitting the 95% load time target
+
+
+==== Script
+
+The percentile rank metric supports scripting.  For example, if our load times
+are in milliseconds but we want to specify values in seconds, we could use
+a script to convert them on-the-fly:
+
+[source,js]
+--------------------------------------------------
+{
+    "aggs" : {
+        "load_time_outlier" : {
+            "percentile_ranks" : {
+                "values" : [3, 5],
+                "script" : "doc['load_time'].value / timeUnit", <1>
+                "params" : {
+                    "timeUnit" : 1000   <2>
+                }
+            }
+        }
+    }
+}
+--------------------------------------------------
+<1> The `field` parameter is replaced with a `script` parameter, which uses the
+script to generate values which percentile ranks are calculated on
+<2> Scripting supports parameterized input just like any other script

src/main/java/org/elasticsearch/search/aggregations/AggregationBuilders.java

@@ -39,6 +39,7 @@ import org.elasticsearch.search.aggregations.metrics.geobounds.GeoBoundsBuilder;
 import org.elasticsearch.search.aggregations.metrics.max.MaxBuilder;
 import org.elasticsearch.search.aggregations.metrics.min.MinBuilder;
 import org.elasticsearch.search.aggregations.metrics.percentiles.PercentilesBuilder;
+import org.elasticsearch.search.aggregations.metrics.percentiles.PercentileRanksBuilder;
 import org.elasticsearch.search.aggregations.metrics.stats.StatsBuilder;
 import org.elasticsearch.search.aggregations.metrics.stats.extended.ExtendedStatsBuilder;
 import org.elasticsearch.search.aggregations.metrics.sum.SumBuilder;
@@ -140,6 +141,10 @@ public class AggregationBuilders {
         return new PercentilesBuilder(name);
     }
 
+    public static PercentileRanksBuilder percentileRanks(String name) {
+        return new PercentileRanksBuilder(name);
+    }
+
     public static CardinalityBuilder cardinality(String name) {
         return new CardinalityBuilder(name);
     }

src/main/java/org/elasticsearch/search/aggregations/AggregationModule.java

@@ -42,6 +42,7 @@ import org.elasticsearch.search.aggregations.metrics.geobounds.GeoBoundsParser;
 import org.elasticsearch.search.aggregations.metrics.max.MaxParser;
 import org.elasticsearch.search.aggregations.metrics.min.MinParser;
 import org.elasticsearch.search.aggregations.metrics.percentiles.PercentilesParser;
+import org.elasticsearch.search.aggregations.metrics.percentiles.PercentileRanksParser;
 import org.elasticsearch.search.aggregations.metrics.stats.StatsParser;
 import org.elasticsearch.search.aggregations.metrics.stats.extended.ExtendedStatsParser;
 import org.elasticsearch.search.aggregations.metrics.sum.SumParser;
@@ -65,6 +66,7 @@ public class AggregationModule extends AbstractModule {
         parsers.add(ExtendedStatsParser.class);
         parsers.add(ValueCountParser.class);
         parsers.add(PercentilesParser.class);
+        parsers.add(PercentileRanksParser.class);
         parsers.add(CardinalityParser.class);
 
         parsers.add(GlobalParser.class);

src/main/java/org/elasticsearch/search/aggregations/TransportAggregationModule.java

@@ -45,6 +45,7 @@ import org.elasticsearch.search.aggregations.metrics.geobounds.InternalGeoBounds
 import org.elasticsearch.search.aggregations.metrics.max.InternalMax;
 import org.elasticsearch.search.aggregations.metrics.min.InternalMin;
 import org.elasticsearch.search.aggregations.metrics.percentiles.InternalPercentiles;
+import org.elasticsearch.search.aggregations.metrics.percentiles.InternalPercentileRanks;
 import org.elasticsearch.search.aggregations.metrics.stats.InternalStats;
 import org.elasticsearch.search.aggregations.metrics.stats.extended.InternalExtendedStats;
 import org.elasticsearch.search.aggregations.metrics.sum.InternalSum;
@@ -67,6 +68,7 @@ public class TransportAggregationModule extends AbstractModule {
         InternalExtendedStats.registerStreams();
         InternalValueCount.registerStreams();
         InternalPercentiles.registerStreams();
+        InternalPercentileRanks.registerStreams();
         InternalCardinality.registerStreams();
 
         // buckets

src/main/java/org/elasticsearch/search/aggregations/metrics/percentiles/AbstractInternalPercentiles.java

@@ -0,0 +1,137 @@
+/*
+ * 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.metrics.percentiles;
+
+import org.elasticsearch.ElasticsearchIllegalArgumentException;
+import org.elasticsearch.Version;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.search.aggregations.InternalAggregation;
+import org.elasticsearch.search.aggregations.metrics.InternalNumericMetricsAggregation;
+import org.elasticsearch.search.aggregations.metrics.percentiles.tdigest.TDigestState;
+import org.elasticsearch.search.aggregations.support.format.ValueFormatterStreams;
+
+import java.io.IOException;
+import java.util.List;
+
+abstract class AbstractInternalPercentiles extends InternalNumericMetricsAggregation.MultiValue {
+
+    protected double[] keys;
+    protected TDigestState state;
+    private boolean keyed;
+
+    AbstractInternalPercentiles() {} // for serialization
+
+    public AbstractInternalPercentiles(String name, double[] keys, TDigestState state, boolean keyed) {
+        super(name);
+        this.keys = keys;
+        this.state = state;
+        this.keyed = keyed;
+    }
+
+    @Override
+    public double value(String name) {
+        return value(Double.parseDouble(name));
+    }
+    
+    public abstract double value(double key);
+
+    @Override
+    public AbstractInternalPercentiles reduce(ReduceContext reduceContext) {
+        List<InternalAggregation> aggregations = reduceContext.aggregations();
+        TDigestState merged = null;
+        for (InternalAggregation aggregation : aggregations) {
+            final AbstractInternalPercentiles percentiles = (AbstractInternalPercentiles) aggregation;
+            if (merged == null) {
+                merged = new TDigestState(percentiles.state.compression());
+            }
+            merged.add(percentiles.state);
+        }
+        return createReduced(getName(), keys, merged, keyed);
+    }
+
+    protected abstract AbstractInternalPercentiles createReduced(String name, double[] keys, TDigestState merged, boolean keyed);
+
+    @Override
+    public void readFrom(StreamInput in) throws IOException {
+        name = in.readString();
+        valueFormatter = ValueFormatterStreams.readOptional(in);
+        if (in.getVersion().before(Version.V_1_2_0)) {
+            final byte id = in.readByte();
+            if (id != 0) {
+                throw new ElasticsearchIllegalArgumentException("Unexpected percentiles aggregator id [" + id + "]");
+            }
+        }
+        keys = new double[in.readInt()];
+        for (int i = 0; i < keys.length; ++i) {
+            keys[i] = in.readDouble();
+        }
+        state = TDigestState.read(in);
+        keyed = in.readBoolean();
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeString(name);
+        ValueFormatterStreams.writeOptional(valueFormatter, out);
+        if (out.getVersion().before(Version.V_1_2_0)) {
+            out.writeByte((byte) 0);
+        }
+        out.writeInt(keys.length);
+        for (int i = 0 ; i < keys.length; ++i) {
+            out.writeDouble(keys[i]);
+        }
+        TDigestState.write(state, out);
+        out.writeBoolean(keyed);
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject(name);
+        if (keyed) {
+            builder.startObject(CommonFields.VALUES);
+            for(int i = 0; i < keys.length; ++i) {
+                String key = String.valueOf(keys[i]);
+                double value = value(keys[i]);
+                builder.field(key, value);
+                if (valueFormatter != null) {
+                    builder.field(key + "_as_string", valueFormatter.format(value));
+                }
+            }
+            builder.endObject();
+        } else {
+            builder.startArray(CommonFields.VALUES);
+            for (int i = 0; i < keys.length; i++) {
+                double value = value(keys[i]);
+                builder.startObject();
+                builder.field(CommonFields.KEY, keys[i]);
+                builder.field(CommonFields.VALUE, value);
+                if (valueFormatter != null) {
+                    builder.field(CommonFields.VALUE_AS_STRING, valueFormatter.format(value));
+                }
+                builder.endObject();
+            }
+            builder.endArray();
+        }
+        builder.endObject();
+        return builder;
+    }
+}

src/main/java/org/elasticsearch/search/aggregations/metrics/percentiles/AbstractPercentilesAggregator.java

@@ -0,0 +1,102 @@
+/*
+ * 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.metrics.percentiles;
+
+import org.apache.lucene.index.AtomicReaderContext;
+import org.elasticsearch.common.lease.Releasables;
+import org.elasticsearch.common.util.ArrayUtils;
+import org.elasticsearch.common.util.ObjectArray;
+import org.elasticsearch.index.fielddata.DoubleValues;
+import org.elasticsearch.search.aggregations.Aggregator;
+import org.elasticsearch.search.aggregations.metrics.NumericMetricsAggregator;
+import org.elasticsearch.search.aggregations.metrics.percentiles.tdigest.TDigestState;
+import org.elasticsearch.search.aggregations.support.AggregationContext;
+import org.elasticsearch.search.aggregations.support.ValuesSource;
+
+import java.io.IOException;
+
+public abstract class AbstractPercentilesAggregator extends NumericMetricsAggregator.MultiValue {
+
+    private static int indexOfKey(double[] keys, double key) {
+        return ArrayUtils.binarySearch(keys, key, 0.001);
+    }
+
+    protected final double[] keys;
+    protected final ValuesSource.Numeric valuesSource;
+    private DoubleValues values;
+    protected ObjectArray<TDigestState> states;
+    protected final double compression;
+    protected final boolean keyed;
+
+    public AbstractPercentilesAggregator(String name, long estimatedBucketsCount, ValuesSource.Numeric valuesSource, AggregationContext context,
+                                 Aggregator parent, double[] keys, double compression, boolean keyed) {
+        super(name, estimatedBucketsCount, context, parent);
+        this.valuesSource = valuesSource;
+        this.keyed = keyed;
+        this.states = bigArrays.newObjectArray(estimatedBucketsCount);
+        this.keys = keys;
+        this.compression = compression;
+    }
+
+    @Override
+    public boolean shouldCollect() {
+        return valuesSource != null;
+    }
+
+    @Override
+    public void setNextReader(AtomicReaderContext reader) {
+        values = valuesSource.doubleValues();
+    }
+
+    @Override
+    public void collect(int doc, long bucketOrd) throws IOException {
+        states = bigArrays.grow(states, bucketOrd + 1);
+    
+        TDigestState state = states.get(bucketOrd);
+        if (state == null) {
+            state = new TDigestState(compression);
+            states.set(bucketOrd, state);
+        }
+    
+        final int valueCount = values.setDocument(doc);
+        for (int i = 0; i < valueCount; i++) {
+            state.add(values.nextValue());
+        }
+    }
+
+    @Override
+    public boolean hasMetric(String name) {
+        return indexOfKey(keys, Double.parseDouble(name)) >= 0;
+    }
+
+    protected TDigestState getState(long bucketOrd) {
+        if (bucketOrd >= states.size()) {
+            return null;
+        }
+        final TDigestState state = states.get(bucketOrd);
+        return state;
+    }
+
+    @Override
+    protected void doClose() {
+        Releasables.close(states);
+    }
+
+}

src/main/java/org/elasticsearch/search/aggregations/metrics/percentiles/AbstractPercentilesParser.java

@@ -0,0 +1,96 @@
+/*
+ * 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.metrics.percentiles;
+
+import com.carrotsearch.hppc.DoubleArrayList;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.search.SearchParseException;
+import org.elasticsearch.search.aggregations.Aggregator;
+import org.elasticsearch.search.aggregations.AggregatorFactory;
+import org.elasticsearch.search.aggregations.support.ValuesSource;
+import org.elasticsearch.search.aggregations.support.ValuesSource.Numeric;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
+import org.elasticsearch.search.aggregations.support.ValuesSourceParser;
+import org.elasticsearch.search.internal.SearchContext;
+
+import java.io.IOException;
+import java.util.Arrays;
+
+public abstract class AbstractPercentilesParser implements Aggregator.Parser {
+
+    public AbstractPercentilesParser() {
+        super();
+    }
+
+    @Override
+    public AggregatorFactory parse(String aggregationName, XContentParser parser, SearchContext context) throws IOException {
+    
+        ValuesSourceParser<ValuesSource.Numeric> vsParser = ValuesSourceParser.numeric(aggregationName, InternalPercentiles.TYPE, context)
+                .requiresSortedValues(true)
+                .build();
+    
+        double[] keys = null;
+        boolean keyed = true;
+        double compression = 100;
+    
+        XContentParser.Token token;
+        String currentFieldName = null;
+        while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
+            if (token == XContentParser.Token.FIELD_NAME) {
+                currentFieldName = parser.currentName();
+            } else if (vsParser.token(currentFieldName, token, parser)) {
+                continue;
+            } else if (token == XContentParser.Token.START_ARRAY) {
+                if (keysFieldName().equals(currentFieldName)) {
+                    DoubleArrayList values = new DoubleArrayList(10);
+                    while ((token = parser.nextToken()) != XContentParser.Token.END_ARRAY) {
+                        double value = parser.doubleValue();
+                        values.add(value);
+                    }
+                    keys = values.toArray();
+                    Arrays.sort(keys);
+                } else {
+                    throw new SearchParseException(context, "Unknown key for a " + token + " in [" + aggregationName + "]: [" + currentFieldName + "].");
+                }
+            } else if (token == XContentParser.Token.VALUE_BOOLEAN) {
+                if ("keyed".equals(currentFieldName)) {
+                    keyed = parser.booleanValue();
+                } else {
+                    throw new SearchParseException(context, "Unknown key for a " + token + " in [" + aggregationName + "]: [" + currentFieldName + "].");
+                }
+            } else if (token == XContentParser.Token.VALUE_NUMBER) {
+                if ("compression".equals(currentFieldName)) {
+                    compression = parser.doubleValue();
+                } else {
+                    throw new SearchParseException(context, "Unknown key for a " + token + " in [" + aggregationName + "]: [" + currentFieldName + "].");
+                }
+            } else {
+                throw new SearchParseException(context, "Unexpected token " + token + " in [" + aggregationName + "].");
+            }
+        }
+        return buildFactory(context, aggregationName, vsParser.config(), keys, compression, keyed);
+    }
+
+    protected abstract AggregatorFactory buildFactory(SearchContext context, String aggregationName, ValuesSourceConfig<Numeric> config, double[] cdfValues,
+            double compression, boolean keyed);
+
+    protected abstract String keysFieldName();
+
+}

src/main/java/org/elasticsearch/search/aggregations/metrics/percentiles/InternalPercentile.java

@@ -0,0 +1,41 @@
+/*
+ * 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.metrics.percentiles;
+
+class InternalPercentile implements Percentile {
+
+    private final double percent;
+    private final double value;
+
+    InternalPercentile(double percent, double value) {
+        this.percent = percent;
+        this.value = value;
+    }
+
+    @Override
+    public double getPercent() {
+        return percent;
+    }
+
+    @Override
+    public double getValue() {
+        return value;
+    }
+}

src/main/java/org/elasticsearch/search/aggregations/metrics/percentiles/InternalPercentileRanks.java

@@ -0,0 +1,114 @@
+/*
+ * 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.metrics.percentiles;
+
+import com.google.common.collect.UnmodifiableIterator;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.search.aggregations.AggregationStreams;
+import org.elasticsearch.search.aggregations.metrics.percentiles.tdigest.TDigestState;
+
+import java.io.IOException;
+import java.util.Iterator;
+
+/**
+*
+*/
+public class InternalPercentileRanks extends AbstractInternalPercentiles implements PercentileRanks {
+
+    public final static Type TYPE = new Type("percentile_ranks");
+
+    public final static AggregationStreams.Stream STREAM = new AggregationStreams.Stream() {
+        @Override
+        public InternalPercentileRanks readResult(StreamInput in) throws IOException {
+            InternalPercentileRanks result = new InternalPercentileRanks();
+            result.readFrom(in);
+            return result;
+        }
+    };
+
+    public static void registerStreams() {
+        AggregationStreams.registerStream(STREAM, TYPE.stream());
+    }
+    
+    InternalPercentileRanks() {} // for serialization
+
+    public InternalPercentileRanks(String name, double[] cdfValues, TDigestState state, boolean keyed) {
+        super(name, cdfValues, state, keyed);
+    }
+
+    @Override
+    public Iterator<Percentile> iterator() {
+        return new Iter(keys, state);
+    }
+
+    @Override
+    public double percent(double value) {
+        return percentileRank(state, value);
+    }
+
+    @Override
+    public double value(double key) {
+        return percent(key);
+    }
+
+    protected AbstractInternalPercentiles createReduced(String name, double[] keys, TDigestState merged, boolean keyed) {
+        return new InternalPercentileRanks(name, keys, merged, keyed);
+    }
+
+    @Override
+    public Type type() {
+        return TYPE;
+    }
+    
+    static double percentileRank(TDigestState state, double value) {
+        double percentileRank = state.cdf(value);
+        if (percentileRank < 0) {
+            percentileRank = 0;
+        }
+        else if (percentileRank > 1) {
+            percentileRank = 1;
+        }
+        return percentileRank * 100;
+    }
+
+    public static class Iter extends UnmodifiableIterator<Percentile> {
+
+        private final double[] values;
+        private final TDigestState state;
+        private int i;
+
+        public Iter(double[] values, TDigestState state) {
+            this.values = values;
+            this.state = state;
+            i = 0;
+        }
+
+        @Override
+        public boolean hasNext() {
+            return i < values.length;
+        }
+
+        @Override
+        public Percentile next() {
+            final Percentile next = new InternalPercentile(percentileRank(state, values[i]), values[i]);
+            ++i;
+            return next;
+        }
+    }
+}

src/main/java/org/elasticsearch/search/aggregations/metrics/percentiles/InternalPercentiles.java

@@ -19,16 +19,9 @@
 package org.elasticsearch.search.aggregations.metrics.percentiles;
 
 import com.google.common.collect.UnmodifiableIterator;
-import org.elasticsearch.ElasticsearchIllegalArgumentException;
-import org.elasticsearch.Version;
 import org.elasticsearch.common.io.stream.StreamInput;
-import org.elasticsearch.common.io.stream.StreamOutput;
-import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.search.aggregations.AggregationStreams;
-import org.elasticsearch.search.aggregations.InternalAggregation;
-import org.elasticsearch.search.aggregations.metrics.InternalNumericMetricsAggregation;
 import org.elasticsearch.search.aggregations.metrics.percentiles.tdigest.TDigestState;
-import org.elasticsearch.search.aggregations.support.format.ValueFormatterStreams;
 
 import java.io.IOException;
 import java.util.Iterator;
@@ -36,7 +29,7 @@ import java.util.Iterator;
 /**
 *
 */
-public class InternalPercentiles extends InternalNumericMetricsAggregation.MultiValue implements Percentiles {
+public class InternalPercentiles extends AbstractInternalPercentiles implements Percentiles {
 
     public final static Type TYPE = new Type("percentiles");
 
@@ -53,27 +46,15 @@ public class InternalPercentiles extends InternalNumericMetricsAggregation.Multi
         AggregationStreams.registerStream(STREAM, TYPE.stream());
     }
 
-    private double[] percents;
-    private TDigestState state;
-    private boolean keyed;
-
     InternalPercentiles() {} // for serialization
 
     public InternalPercentiles(String name, double[] percents, TDigestState state, boolean keyed) {
-        super(name);
-        this.percents = percents;
-        this.state = state;
-        this.keyed = keyed;
-    }
-
-    @Override
-    public double value(String name) {
-        return percentile(Double.parseDouble(name));
+        super(name, percents, state, keyed);
     }
 
     @Override
-    public Type type() {
-        return TYPE;
+    public Iterator<Percentile> iterator() {
+        return new Iter(keys, state);
     }
 
     @Override
@@ -82,89 +63,20 @@ public class InternalPercentiles extends InternalNumericMetricsAggregation.Multi
     }
 
     @Override
-    public Iterator<Percentiles.Percentile> iterator() {
-        return new Iter(percents, state);
-    }
-
-    @Override
-    public InternalPercentiles reduce(ReduceContext reduceContext) {
-        TDigestState merged = null;
-        for (InternalAggregation aggregation : reduceContext.aggregations()) {
-            final InternalPercentiles percentiles = (InternalPercentiles) aggregation;
-            if (merged == null) {
-                merged = new TDigestState(percentiles.state.compression());
-            }
-            merged.add(percentiles.state);
-        }
-        return new InternalPercentiles(getName(), percents, merged, keyed);
-    }
-
-    @Override
-    public void readFrom(StreamInput in) throws IOException {
-        name = in.readString();
-        valueFormatter = ValueFormatterStreams.readOptional(in);
-        if (in.getVersion().before(Version.V_1_2_0)) {
-            final byte id = in.readByte();
-            if (id != 0) {
-                throw new ElasticsearchIllegalArgumentException("Unexpected percentiles aggregator id [" + id + "]");
-            }
-        }
-        percents = new double[in.readInt()];
-        for (int i = 0; i < percents.length; ++i) {
-            percents[i] = in.readDouble();
-        }
-        state = TDigestState.read(in);
-        keyed = in.readBoolean();
+    public double value(double key) {
+        return percentile(key);
     }
 
-    @Override
-    public void writeTo(StreamOutput out) throws IOException {
-        out.writeString(name);
-        ValueFormatterStreams.writeOptional(valueFormatter, out);
-        if (out.getVersion().before(Version.V_1_2_0)) {
-            out.writeByte((byte) 0);
-        }
-        out.writeInt(percents.length);
-        for (int i = 0 ; i < percents.length; ++i) {
-            out.writeDouble(percents[i]);
-        }
-        TDigestState.write(state, out);
-        out.writeBoolean(keyed);
+    protected AbstractInternalPercentiles createReduced(String name, double[] keys, TDigestState merged, boolean keyed) {
+        return new InternalPercentiles(name, keys, merged, keyed);
     }
 
     @Override
-    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-        builder.startObject(name);
-        if (keyed) {
-            builder.startObject(CommonFields.VALUES);
-            for(int i = 0; i < percents.length; ++i) {
-                String key = String.valueOf(percents[i]);
-                double value = percentile(percents[i]);
-                builder.field(key, value);
-                if (valueFormatter != null) {
-                    builder.field(key + "_as_string", valueFormatter.format(value));
-                }
-            }
-            builder.endObject();
-        } else {
-            builder.startArray(CommonFields.VALUES);
-            for (int i = 0; i < percents.length; i++) {
-                double value = percentile(percents[i]);
-                builder.startObject();
-                builder.field(CommonFields.KEY, percents[i]);
-                builder.field(CommonFields.VALUE, value);
-                if (valueFormatter != null) {
-                    builder.field(CommonFields.VALUE_AS_STRING, valueFormatter.format(value));
-                }
-                builder.endObject();
-            }
-            builder.endArray();
-        }
-        builder.endObject();
-        return builder;
+    public Type type() {
+        return TYPE;
     }
 
-    public static class Iter extends UnmodifiableIterator<Percentiles.Percentile> {
+    public static class Iter extends UnmodifiableIterator<Percentile> {
 
         private final double[] percents;
         private final TDigestState state;
@@ -182,31 +94,10 @@ public class InternalPercentiles extends InternalNumericMetricsAggregation.Multi
         }
 
         @Override
-        public Percentiles.Percentile next() {
-            final Percentiles.Percentile next = new InnerPercentile(percents[i], state.quantile(percents[i] / 100));
+        public Percentile next() {
+            final Percentile next = new InternalPercentile(percents[i], state.quantile(percents[i] / 100));
             ++i;
             return next;
         }
     }
-
-    private static class InnerPercentile implements Percentiles.Percentile {
-
-        private final double percent;
-        private final double value;
-
-        private InnerPercentile(double percent, double value) {
-            this.percent = percent;
-            this.value = value;
-        }
-
-        @Override
-        public double getPercent() {
-            return percent;
-        }
-
-        @Override
-        public double getValue() {
-            return value;
-        }
-    }
 }

src/main/java/org/elasticsearch/search/aggregations/metrics/percentiles/Percentile.java

@@ -0,0 +1,28 @@
+/*
+ * 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.metrics.percentiles;
+
+public interface Percentile {
+
+    double getPercent();
+
+    double getValue();
+
+}

src/main/java/org/elasticsearch/search/aggregations/metrics/percentiles/PercentileRanks.java

@@ -0,0 +1,27 @@
+/*
+ * 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.metrics.percentiles;
+
+import org.elasticsearch.search.aggregations.Aggregation;
+
+public interface PercentileRanks extends Aggregation, Iterable<Percentile>{
+
+    double percent(double percentile);
+}

src/main/java/org/elasticsearch/search/aggregations/metrics/percentiles/PercentileRanksAggregator.java

@@ -0,0 +1,86 @@
+/*
+ * 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.metrics.percentiles;
+
+import org.elasticsearch.search.aggregations.Aggregator;
+import org.elasticsearch.search.aggregations.InternalAggregation;
+import org.elasticsearch.search.aggregations.metrics.percentiles.tdigest.TDigestState;
+import org.elasticsearch.search.aggregations.support.*;
+import org.elasticsearch.search.aggregations.support.ValuesSource.Numeric;
+
+/**
+ *
+ */
+public class PercentileRanksAggregator extends AbstractPercentilesAggregator {
+
+    public PercentileRanksAggregator(String name, long estimatedBucketsCount, Numeric valuesSource, AggregationContext context,
+            Aggregator parent, double[] percents, double compression, boolean keyed) {
+        super(name, estimatedBucketsCount, valuesSource, context, parent, percents, compression, keyed);
+    }
+
+    @Override
+    public InternalAggregation buildAggregation(long owningBucketOrdinal) {
+        TDigestState state = getState(owningBucketOrdinal);
+        if (state == null) {
+            return buildEmptyAggregation();
+        } else {
+            return new InternalPercentileRanks(name, keys, state, keyed);
+        }
+    }
+
+    @Override
+    public InternalAggregation buildEmptyAggregation() {
+        return new InternalPercentileRanks(name, keys, new TDigestState(compression), keyed);
+    }
+
+    @Override
+    public double metric(String name, long bucketOrd) {
+        TDigestState state = getState(bucketOrd);
+        if (state == null) {
+            return Double.NaN;
+        } else {
+            return InternalPercentileRanks.percentileRank(state, Double.valueOf(name));
+        }
+    }
+
+    public static class Factory extends ValuesSourceAggregatorFactory.LeafOnly<ValuesSource.Numeric> {
+
+        private final double[] values;
+        private final double compression;
+        private final boolean keyed;
+
+        public Factory(String name, ValuesSourceConfig<ValuesSource.Numeric> valuesSourceConfig,
+                double[] values, double compression, boolean keyed) {
+            super(name, InternalPercentiles.TYPE.name(), valuesSourceConfig);
+            this.values = values;
+            this.compression = compression;
+            this.keyed = keyed;
+        }
+
+        @Override
+        protected Aggregator createUnmapped(AggregationContext aggregationContext, Aggregator parent) {
+            return new PercentileRanksAggregator(name, 0, null, aggregationContext, parent, values, compression, keyed);
+        }
+
+        @Override
+        protected Aggregator create(ValuesSource.Numeric valuesSource, long expectedBucketsCount, AggregationContext aggregationContext, Aggregator parent) {
+            return new PercentileRanksAggregator(name, expectedBucketsCount, valuesSource, aggregationContext, parent, values, compression, keyed);
+        }
+    }
+}

src/main/java/org/elasticsearch/search/aggregations/metrics/percentiles/PercentileRanksBuilder.java

@@ -0,0 +1,60 @@
+/*
+ * 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.metrics.percentiles;
+
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.search.aggregations.metrics.ValuesSourceMetricsAggregationBuilder;
+
+import java.io.IOException;
+
+/**
+ *
+ */
+public class PercentileRanksBuilder extends ValuesSourceMetricsAggregationBuilder<PercentileRanksBuilder> {
+
+    private double[] values;
+    private Double compression;
+
+    public PercentileRanksBuilder(String name) {
+        super(name, InternalPercentileRanks.TYPE.name());
+    }
+
+    public PercentileRanksBuilder percentiles(double... values) {
+        this.values = values;
+        return this;
+    }
+
+    public PercentileRanksBuilder compression(double compression) {
+        this.compression = compression;
+        return this;
+    }
+
+    @Override
+    protected void internalXContent(XContentBuilder builder, Params params) throws IOException {
+        super.internalXContent(builder, params);
+
+        if (values != null) {
+            builder.field("values", values);
+        }
+
+        if (compression != null) {
+            builder.field("compression", compression);
+        }
+    }
+}

src/main/java/org/elasticsearch/search/aggregations/metrics/percentiles/PercentileRanksParser.java

@@ -0,0 +1,48 @@
+/*
+ * 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.metrics.percentiles;
+
+import org.elasticsearch.search.SearchParseException;
+import org.elasticsearch.search.aggregations.AggregatorFactory;
+import org.elasticsearch.search.aggregations.support.ValuesSource.Numeric;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
+import org.elasticsearch.search.internal.SearchContext;
+
+/**
+ *
+ */
+public class PercentileRanksParser extends AbstractPercentilesParser {
+
+    @Override
+    public String type() {
+        return InternalPercentileRanks.TYPE.name();
+    }
+
+    protected String keysFieldName() {
+        return "values";
+    }
+    
+    protected AggregatorFactory buildFactory(SearchContext context, String aggregationName, ValuesSourceConfig<Numeric> valuesSourceConfig, double[] keys, double compression, boolean keyed) {
+        if (keys == null) {
+            throw new SearchParseException(context, "Missing token values in [" + aggregationName + "].");
+        }
+        return new PercentileRanksAggregator.Factory(aggregationName, valuesSourceConfig, keys, compression, keyed);
+    }
+
+}

src/main/java/org/elasticsearch/search/aggregations/metrics/percentiles/Percentiles.java

@@ -23,15 +23,7 @@ import org.elasticsearch.search.aggregations.Aggregation;
 /**
  *
  */
-public interface Percentiles extends Aggregation, Iterable<Percentiles.Percentile> {
-
-    public static interface Percentile {
-
-        double getPercent();
-
-        double getValue();
-
-    }
+public interface Percentiles extends Aggregation, Iterable<Percentile> {
 
     double percentile(double percent);
 

src/main/java/org/elasticsearch/search/aggregations/metrics/percentiles/PercentilesAggregator.java

@@ -18,89 +18,32 @@
  */
 package org.elasticsearch.search.aggregations.metrics.percentiles;
 
-import org.apache.lucene.index.AtomicReaderContext;
-import org.elasticsearch.common.lease.Releasables;
-import org.elasticsearch.common.util.ArrayUtils;
-import org.elasticsearch.common.util.ObjectArray;
-import org.elasticsearch.index.fielddata.DoubleValues;
 import org.elasticsearch.search.aggregations.Aggregator;
 import org.elasticsearch.search.aggregations.InternalAggregation;
-import org.elasticsearch.search.aggregations.metrics.NumericMetricsAggregator;
 import org.elasticsearch.search.aggregations.metrics.percentiles.tdigest.TDigestState;
-import org.elasticsearch.search.aggregations.support.AggregationContext;
-import org.elasticsearch.search.aggregations.support.ValuesSource;
-import org.elasticsearch.search.aggregations.support.ValuesSourceAggregatorFactory;
-import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
-
-import java.io.IOException;
+import org.elasticsearch.search.aggregations.support.*;
+import org.elasticsearch.search.aggregations.support.ValuesSource.Numeric;
 
 /**
  *
  */
-public class PercentilesAggregator extends NumericMetricsAggregator.MultiValue {
-
-    private static int indexOfPercent(double[] percents, double percent) {
-        return ArrayUtils.binarySearch(percents, percent, 0.001);
-    }
-
-    private final double[] percents;
-    private final ValuesSource.Numeric valuesSource;
-    private DoubleValues values;
-
-    private ObjectArray<TDigestState> states;
-    private final double compression;
-    private final boolean keyed;
+public class PercentilesAggregator extends AbstractPercentilesAggregator {
 
-
-    public PercentilesAggregator(String name, long estimatedBucketsCount, ValuesSource.Numeric valuesSource, AggregationContext context,
-                                 Aggregator parent, double[] percents, double compression, boolean keyed) {
-        super(name, estimatedBucketsCount, context, parent);
-        this.valuesSource = valuesSource;
-        this.keyed = keyed;
-        this.states = bigArrays.newObjectArray(estimatedBucketsCount);
-        this.percents = percents;
-        this.compression = compression;
+    public PercentilesAggregator(String name, long estimatedBucketsCount, Numeric valuesSource, AggregationContext context,
+            Aggregator parent, double[] percents, double compression, boolean keyed) {
+        super(name, estimatedBucketsCount, valuesSource, context, parent, percents, compression, keyed);
     }
 
     @Override
-    public boolean shouldCollect() {
-        return valuesSource != null;
-    }
-
-    @Override
-    public void setNextReader(AtomicReaderContext reader) {
-        values = valuesSource.doubleValues();
-    }
-
-    @Override
-    public void collect(int doc, long bucketOrd) throws IOException {
-        states = bigArrays.grow(states, bucketOrd + 1);
-
-        TDigestState state = states.get(bucketOrd);
+    public InternalAggregation buildAggregation(long owningBucketOrdinal) {
+        TDigestState state = getState(owningBucketOrdinal);
         if (state == null) {
-            state = new TDigestState(compression);
-            states.set(bucketOrd, state);
-        }
-
-        final int valueCount = values.setDocument(doc);
-        for (int i = 0; i < valueCount; i++) {
-            state.add(values.nextValue());
-        }
-    }
-
-    @Override
-    public boolean hasMetric(String name) {
-        return indexOfPercent(percents, Double.parseDouble(name)) >= 0;
-    }
-
-    private TDigestState getState(long bucketOrd) {
-        if (bucketOrd >= states.size()) {
-            return null;
+            return buildEmptyAggregation();
+        } else {
+            return new InternalPercentiles(name, keys, state, keyed);
         }
-        final TDigestState state = states.get(bucketOrd);
-        return state;
     }
-
+    
     @Override
     public double metric(String name, long bucketOrd) {
         TDigestState state = getState(bucketOrd);
@@ -111,24 +54,9 @@ public class PercentilesAggregator extends NumericMetricsAggregator.MultiValue {
         }
     }
 
-    @Override
-    public InternalAggregation buildAggregation(long owningBucketOrdinal) {
-        TDigestState state = getState(owningBucketOrdinal);
-        if (state == null) {
-            return buildEmptyAggregation();
-        } else {
-            return new InternalPercentiles(name, percents, state, keyed);
-        }
-    }
-
     @Override
     public InternalAggregation buildEmptyAggregation() {
-        return new InternalPercentiles(name, percents, new TDigestState(compression), keyed);
-    }
-
-    @Override
-    protected void doClose() {
-        Releasables.close(states);
+        return new InternalPercentiles(name, keys, new TDigestState(compression), keyed);
     }
 
     public static class Factory extends ValuesSourceAggregatorFactory.LeafOnly<ValuesSource.Numeric> {

src/main/java/org/elasticsearch/search/aggregations/metrics/percentiles/PercentilesParser.java

@@ -18,22 +18,15 @@
  */
 package org.elasticsearch.search.aggregations.metrics.percentiles;
 
-import com.carrotsearch.hppc.DoubleArrayList;
-import org.elasticsearch.common.xcontent.XContentParser;
-import org.elasticsearch.search.SearchParseException;
-import org.elasticsearch.search.aggregations.Aggregator;
 import org.elasticsearch.search.aggregations.AggregatorFactory;
-import org.elasticsearch.search.aggregations.support.ValuesSource;
-import org.elasticsearch.search.aggregations.support.ValuesSourceParser;
+import org.elasticsearch.search.aggregations.support.ValuesSource.Numeric;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.internal.SearchContext;
 
-import java.io.IOException;
-import java.util.Arrays;
-
 /**
  *
  */
-public class PercentilesParser implements Aggregator.Parser {
+public class PercentilesParser extends AbstractPercentilesParser {
 
     private final static double[] DEFAULT_PERCENTS = new double[] { 1, 5, 25, 50, 75, 95, 99 };
 
@@ -42,63 +35,15 @@ public class PercentilesParser implements Aggregator.Parser {
         return InternalPercentiles.TYPE.name();
     }
 
-    /**
-     * We must override the parse method because we need to allow custom parameters
-     * (execution_hint, etc) which is not possible otherwise
-     */
-    @Override
-    public AggregatorFactory parse(String aggregationName, XContentParser parser, SearchContext context) throws IOException {
-
-        ValuesSourceParser<ValuesSource.Numeric> vsParser = ValuesSourceParser.numeric(aggregationName, InternalPercentiles.TYPE, context)
-                .requiresSortedValues(true)
-                .build();
-
-        double[] percents = DEFAULT_PERCENTS;
-        boolean keyed = true;
-        double compression = 100;
-
-        XContentParser.Token token;
-        String currentFieldName = null;
-        while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
-            if (token == XContentParser.Token.FIELD_NAME) {
-                currentFieldName = parser.currentName();
-            } else if (vsParser.token(currentFieldName, token, parser)) {
-                continue;
-            } else if (token == XContentParser.Token.START_ARRAY) {
-                if ("percents".equals(currentFieldName)) {
-                    DoubleArrayList values = new DoubleArrayList(10);
-                    while ((token = parser.nextToken()) != XContentParser.Token.END_ARRAY) {
-                        double percent = parser.doubleValue();
-                        if (percent < 0 || percent > 100) {
-                            throw new SearchParseException(context, "the percents in the percentiles aggregation [" +
-                                    aggregationName + "] must be in the [0, 100] range");
-                        }
-                        values.add(percent);
-                    }
-                    percents = values.toArray();
-                    // Some impls rely on the fact that percents are sorted
-                    Arrays.sort(percents);
-                } else {
-                    throw new SearchParseException(context, "Unknown key for a " + token + " in [" + aggregationName + "]: [" + currentFieldName + "].");
-                }
-            } else if (token == XContentParser.Token.VALUE_BOOLEAN) {
-                if ("keyed".equals(currentFieldName)) {
-                    keyed = parser.booleanValue();
-                } else {
-                    throw new SearchParseException(context, "Unknown key for a " + token + " in [" + aggregationName + "]: [" + currentFieldName + "].");
-                }
-            } else if (token == XContentParser.Token.VALUE_NUMBER) {
-                if ("compression".equals(currentFieldName)) {
-                    compression = parser.doubleValue();
-                } else {
-                    throw new SearchParseException(context, "Unknown key for a " + token + " in [" + aggregationName + "]: [" + currentFieldName + "].");
-                }
-            } else {
-                throw new SearchParseException(context, "Unexpected token " + token + " in [" + aggregationName + "].");
-            }
+    protected String keysFieldName() {
+        return "percents";
+    }
+    
+    protected AggregatorFactory buildFactory(SearchContext context, String aggregationName, ValuesSourceConfig<Numeric> valuesSourceConfig, double[] keys, double compression, boolean keyed) {
+        if (keys == null) {
+            keys = DEFAULT_PERCENTS;
         }
-
-        return new PercentilesAggregator.Factory(aggregationName, vsParser.config(), percents, compression, keyed);
+        return new PercentilesAggregator.Factory(aggregationName, valuesSourceConfig, keys, compression, keyed);
     }
 
 }

src/test/java/org/elasticsearch/benchmark/search/aggregations/PercentilesAggregationSearchBenchmark.java

@@ -19,6 +19,8 @@
 
 package org.elasticsearch.benchmark.search.aggregations;
 
+import org.elasticsearch.search.aggregations.metrics.percentiles.Percentile;
+
 import com.google.common.collect.Maps;
 import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
 import org.elasticsearch.action.bulk.BulkRequestBuilder;
@@ -34,7 +36,6 @@ import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.json.JsonXContent;
 import org.elasticsearch.node.Node;
 import org.elasticsearch.search.aggregations.metrics.percentiles.Percentiles;
-import org.elasticsearch.search.aggregations.metrics.percentiles.Percentiles.Percentile;
 
 import java.util.*;
 import java.util.concurrent.TimeUnit;

src/test/java/org/elasticsearch/search/aggregations/metrics/PercentileRanksTests.java

@@ -0,0 +1,405 @@
+/*
+ * 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.metrics;
+
+import com.google.common.collect.Lists;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.common.logging.Loggers;
+import org.elasticsearch.search.aggregations.bucket.histogram.Histogram;
+import org.elasticsearch.search.aggregations.bucket.histogram.Histogram.Order;
+import org.elasticsearch.search.aggregations.metrics.percentiles.Percentile;
+import org.elasticsearch.search.aggregations.metrics.percentiles.PercentileRanks;
+import org.elasticsearch.search.aggregations.metrics.percentiles.PercentileRanksBuilder;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static org.elasticsearch.index.query.QueryBuilders.matchAllQuery;
+import static org.elasticsearch.search.aggregations.AggregationBuilders.histogram;
+import static org.elasticsearch.search.aggregations.AggregationBuilders.percentileRanks;
+import static org.hamcrest.Matchers.*;
+
+/**
+ *
+ */
+public class PercentileRanksTests extends AbstractNumericTests {
+
+    private static double[] randomPercents(long minValue, long maxValue) {
+        
+        final int length = randomIntBetween(1, 20);
+        final double[] percents = new double[length];
+        for (int i = 0; i < percents.length; ++i) {
+            switch (randomInt(20)) {
+            case 0:
+                percents[i] = minValue;
+                break;
+            case 1:
+                percents[i] = maxValue;
+                break;
+            default:
+                percents[i] = (randomDouble() * (maxValue - minValue)) + minValue;
+                break;
+            }
+        }
+        Arrays.sort(percents);
+        Loggers.getLogger(PercentileRanksTests.class).info("Using percentiles={}", Arrays.toString(percents));
+        return percents;
+    }
+
+    private static PercentileRanksBuilder randomCompression(PercentileRanksBuilder builder) {
+        if (randomBoolean()) {
+            builder.compression(randomIntBetween(20, 120) + randomDouble());
+        }
+        return builder;
+    }
+
+    private void assertConsistent(double[] pcts, PercentileRanks percentiles, long minValue, long maxValue) {
+        final List<Percentile> percentileList = Lists.newArrayList(percentiles);
+        assertEquals(pcts.length, percentileList.size());
+        for (int i = 0; i < pcts.length; ++i) {
+            final Percentile percentile = percentileList.get(i);
+            assertThat(percentile.getValue(), equalTo(pcts[i]));
+            assertThat(percentile.getPercent(), greaterThanOrEqualTo(0.0));
+            assertThat(percentile.getPercent(), lessThanOrEqualTo(100.0));
+
+            if (percentile.getPercent() == 0) {
+                assertThat(percentile.getValue(), lessThanOrEqualTo((double) minValue));
+            }
+            if (percentile.getPercent() == 100) {
+                assertThat(percentile.getValue(), greaterThanOrEqualTo((double) maxValue));
+            }
+        }
+
+        for (int i = 1; i < percentileList.size(); ++i) {
+            assertThat(percentileList.get(i).getValue(), greaterThanOrEqualTo(percentileList.get(i - 1).getValue()));
+        }
+    }
+
+    @Test
+    public void testEmptyAggregation() throws Exception {
+
+        SearchResponse searchResponse = client().prepareSearch("empty_bucket_idx")
+                .setQuery(matchAllQuery())
+                .addAggregation(histogram("histo").field("value").interval(1l).minDocCount(0)
+                        .subAggregation(randomCompression(percentileRanks("percentile_ranks"))
+                                .percentiles(10, 15)))
+                .execute().actionGet();
+
+        assertThat(searchResponse.getHits().getTotalHits(), equalTo(2l));
+        Histogram histo = searchResponse.getAggregations().get("histo");
+        assertThat(histo, notNullValue());
+        Histogram.Bucket bucket = histo.getBucketByKey(1l);
+        assertThat(bucket, notNullValue());
+
+        PercentileRanks reversePercentiles = bucket.getAggregations().get("percentile_ranks");
+        assertThat(reversePercentiles, notNullValue());
+        assertThat(reversePercentiles.getName(), equalTo("percentile_ranks"));
+        assertThat(reversePercentiles.percent(10), equalTo(Double.NaN));
+        assertThat(reversePercentiles.percent(15), equalTo(Double.NaN));
+    }
+
+    @Test
+    public void testUnmapped() throws Exception {
+        SearchResponse searchResponse = client().prepareSearch("idx_unmapped")
+                .setQuery(matchAllQuery())
+                .addAggregation(randomCompression(percentileRanks("percentile_ranks"))
+                        .field("value")
+                        .percentiles(0, 10, 15, 100))
+                .execute().actionGet();
+
+        assertThat(searchResponse.getHits().getTotalHits(), equalTo(0l));
+
+        PercentileRanks reversePercentiles = searchResponse.getAggregations().get("percentile_ranks");
+        assertThat(reversePercentiles, notNullValue());
+        assertThat(reversePercentiles.getName(), equalTo("percentile_ranks"));
+        assertThat(reversePercentiles.percent(0), equalTo(Double.NaN));
+        assertThat(reversePercentiles.percent(10), equalTo(Double.NaN));
+        assertThat(reversePercentiles.percent(15), equalTo(Double.NaN));
+        assertThat(reversePercentiles.percent(100), equalTo(Double.NaN));
+    }
+
+    @Test
+    public void testSingleValuedField() throws Exception {
+        final double[] pcts = randomPercents(minValue, maxValue);
+        SearchResponse searchResponse = client().prepareSearch("idx")
+                .setQuery(matchAllQuery())
+                .addAggregation(randomCompression(percentileRanks("percentile_ranks"))
+                        .field("value")
+                        .percentiles(pcts))
+                .execute().actionGet();
+
+        assertThat(searchResponse.getHits().getTotalHits(), equalTo(10l));
+
+        final PercentileRanks percentiles = searchResponse.getAggregations().get("percentile_ranks");
+        assertConsistent(pcts, percentiles, minValue, maxValue);
+    }
+
+    @Test
+    public void testSingleValuedFieldOutsideRange() throws Exception {
+        final double[] pcts = new double[] {minValue - 1, maxValue + 1};
+        SearchResponse searchResponse = client().prepareSearch("idx")
+                .setQuery(matchAllQuery())
+                .addAggregation(randomCompression(percentileRanks("percentile_ranks"))
+                        .field("value")
+                        .percentiles(pcts))
+                .execute().actionGet();
+
+        assertThat(searchResponse.getHits().getTotalHits(), equalTo(10l));
+
+        final PercentileRanks percentiles = searchResponse.getAggregations().get("percentile_ranks");
+        assertConsistent(pcts, percentiles, minValue, maxValue);
+    }
+
+    @Test
+    public void testSingleValuedField_PartiallyUnmapped() throws Exception {
+        final double[] pcts = randomPercents(minValue, maxValue);
+        SearchResponse searchResponse = client().prepareSearch("idx", "idx_unmapped")
+                .setQuery(matchAllQuery())
+                .addAggregation(randomCompression(percentileRanks("percentile_ranks"))
+                        .field("value")
+                        .percentiles(pcts))
+                .execute().actionGet();
+
+        assertThat(searchResponse.getHits().getTotalHits(), equalTo(10l));
+
+        final PercentileRanks percentiles = searchResponse.getAggregations().get("percentile_ranks");
+        assertConsistent(pcts, percentiles, minValue, maxValue);
+    }
+
+    @Test
+    public void testSingleValuedField_WithValueScript() throws Exception {
+        final double[] pcts = randomPercents(minValue - 1, maxValue - 1);
+        SearchResponse searchResponse = client().prepareSearch("idx")
+                .setQuery(matchAllQuery())
+                .addAggregation(randomCompression(percentileRanks("percentile_ranks"))
+                        .field("value").script("_value - 1")
+                        .percentiles(pcts))
+                .execute().actionGet();
+
+        assertThat(searchResponse.getHits().getTotalHits(), equalTo(10l));
+
+        final PercentileRanks percentiles = searchResponse.getAggregations().get("percentile_ranks");
+        assertConsistent(pcts, percentiles, minValue - 1, maxValue - 1);
+    }
+
+    @Test
+    public void testSingleValuedField_WithValueScript_WithParams() throws Exception {
+        final double[] pcts = randomPercents(minValue - 1, maxValue - 1);
+        SearchResponse searchResponse = client().prepareSearch("idx")
+                .setQuery(matchAllQuery())
+                .addAggregation(randomCompression(percentileRanks("percentile_ranks"))
+                        .field("value").script("_value - dec").param("dec", 1)
+                        .percentiles(pcts))
+                .execute().actionGet();
+
+        assertThat(searchResponse.getHits().getTotalHits(), equalTo(10l));
+
+        final PercentileRanks percentiles = searchResponse.getAggregations().get("percentile_ranks");
+        assertConsistent(pcts, percentiles, minValue - 1, maxValue - 1);
+    }
+
+    @Test
+    public void testMultiValuedField() throws Exception {
+        final double[] pcts = randomPercents(minValues, maxValues);
+        SearchResponse searchResponse = client().prepareSearch("idx")
+                .setQuery(matchAllQuery())
+                .addAggregation(randomCompression(percentileRanks("percentile_ranks"))
+                        .field("values")
+                        .percentiles(pcts))
+                .execute().actionGet();
+
+        assertThat(searchResponse.getHits().getTotalHits(), equalTo(10l));
+
+        final PercentileRanks percentiles = searchResponse.getAggregations().get("percentile_ranks");
+        assertConsistent(pcts, percentiles, minValues, maxValues);
+    }
+
+    @Test
+    public void testMultiValuedField_WithValueScript() throws Exception {
+        final double[] pcts = randomPercents(minValues - 1, maxValues - 1);
+        SearchResponse searchResponse = client().prepareSearch("idx")
+                .setQuery(matchAllQuery())
+                .addAggregation(randomCompression(percentileRanks("percentile_ranks"))
+                        .field("values").script("_value - 1")
+                        .percentiles(pcts))
+                .execute().actionGet();
+
+        assertThat(searchResponse.getHits().getTotalHits(), equalTo(10l));
+
+        final PercentileRanks percentiles = searchResponse.getAggregations().get("percentile_ranks");
+        assertConsistent(pcts, percentiles, minValues - 1, maxValues - 1);
+    }
+
+    @Test
+    public void testMultiValuedField_WithValueScript_Reverse() throws Exception {
+        final double[] pcts = randomPercents(-maxValues, -minValues);
+        SearchResponse searchResponse = client().prepareSearch("idx")
+                .setQuery(matchAllQuery())
+                .addAggregation(randomCompression(percentileRanks("percentile_ranks"))
+                        .field("values").script("_value * -1")
+                        .percentiles(pcts))
+                .execute().actionGet();
+
+        assertThat(searchResponse.getHits().getTotalHits(), equalTo(10l));
+
+        final PercentileRanks percentiles = searchResponse.getAggregations().get("percentile_ranks");
+        assertConsistent(pcts, percentiles, -maxValues, -minValues);
+    }
+
+    @Test
+    public void testMultiValuedField_WithValueScript_WithParams() throws Exception {
+        final double[] pcts = randomPercents(minValues - 1, maxValues - 1);
+        SearchResponse searchResponse = client().prepareSearch("idx")
+                .setQuery(matchAllQuery())
+                .addAggregation(randomCompression(percentileRanks("percentile_ranks"))
+                        .field("values").script("_value - dec").param("dec", 1)
+                        .percentiles(pcts))
+                .execute().actionGet();
+
+        assertThat(searchResponse.getHits().getTotalHits(), equalTo(10l));
+
+        final PercentileRanks percentiles = searchResponse.getAggregations().get("percentile_ranks");
+        assertConsistent(pcts, percentiles, minValues - 1, maxValues - 1);
+    }
+
+    @Test
+    public void testScript_SingleValued() throws Exception {
+        final double[] pcts = randomPercents(minValue, maxValue);
+        SearchResponse searchResponse = client().prepareSearch("idx")
+                .setQuery(matchAllQuery())
+                .addAggregation(randomCompression(percentileRanks("percentile_ranks"))
+                        .script("doc['value'].value")
+                        .percentiles(pcts))
+                .execute().actionGet();
+
+        assertThat(searchResponse.getHits().getTotalHits(), equalTo(10l));
+
+        final PercentileRanks percentiles = searchResponse.getAggregations().get("percentile_ranks");
+        assertConsistent(pcts, percentiles, minValue, maxValue);
+    }
+
+    @Test
+    public void testScript_SingleValued_WithParams() throws Exception {
+        final double[] pcts = randomPercents(minValue - 1, maxValue - 1);
+        SearchResponse searchResponse = client().prepareSearch("idx")
+                .setQuery(matchAllQuery())
+                .addAggregation(randomCompression(percentileRanks("percentile_ranks"))
+                        .script("doc['value'].value - dec").param("dec", 1)
+                        .percentiles(pcts))
+                .execute().actionGet();
+
+        assertThat(searchResponse.getHits().getTotalHits(), equalTo(10l));
+
+        final PercentileRanks percentiles = searchResponse.getAggregations().get("percentile_ranks");
+        assertConsistent(pcts, percentiles, minValue - 1, maxValue - 1);
+    }
+
+    @Test
+    public void testScript_ExplicitSingleValued_WithParams() throws Exception {
+        final double[] pcts = randomPercents(minValue -1 , maxValue - 1);
+        SearchResponse searchResponse = client().prepareSearch("idx")
+                .setQuery(matchAllQuery())
+                .addAggregation(randomCompression(percentileRanks("percentile_ranks"))
+                        .script("doc['value'].value - dec").param("dec", 1)
+                        .percentiles(pcts))
+                .execute().actionGet();
+
+        assertThat(searchResponse.getHits().getTotalHits(), equalTo(10l));
+
+        final PercentileRanks percentiles = searchResponse.getAggregations().get("percentile_ranks");
+        assertConsistent(pcts, percentiles, minValue - 1, maxValue - 1);
+    }
+
+    @Test
+    public void testScript_MultiValued() throws Exception {
+        final double[] pcts = randomPercents(minValues, maxValues);
+        SearchResponse searchResponse = client().prepareSearch("idx")
+                .setQuery(matchAllQuery())
+                .addAggregation(randomCompression(percentileRanks("percentile_ranks"))
+                        .script("doc['values'].values")
+                        .percentiles(pcts))
+                .execute().actionGet();
+
+        assertThat(searchResponse.getHits().getTotalHits(), equalTo(10l));
+
+        final PercentileRanks percentiles = searchResponse.getAggregations().get("percentile_ranks");
+        assertConsistent(pcts, percentiles, minValues, maxValues);
+    }
+
+    @Test
+    public void testScript_ExplicitMultiValued() throws Exception {
+        final double[] pcts = randomPercents(minValues, maxValues);
+        SearchResponse searchResponse = client().prepareSearch("idx")
+                .setQuery(matchAllQuery())
+                .addAggregation(randomCompression(percentileRanks("percentile_ranks"))
+                        .script("doc['values'].values")
+                        .percentiles(pcts))
+                .execute().actionGet();
+
+        assertThat(searchResponse.getHits().getTotalHits(), equalTo(10l));
+
+        final PercentileRanks percentiles = searchResponse.getAggregations().get("percentile_ranks");
+        assertConsistent(pcts, percentiles, minValues, maxValues);
+    }
+
+    @Test
+    public void testScript_MultiValued_WithParams() throws Exception {
+        final double[] pcts = randomPercents(minValues - 1, maxValues - 1);
+        SearchResponse searchResponse = client().prepareSearch("idx")
+                .setQuery(matchAllQuery())
+                .addAggregation(randomCompression(percentileRanks("percentile_ranks"))
+                        .script("List values = doc['values'].values; double[] res = new double[values.length]; for (int i = 0; i < res.length; i++) { res[i] = values.get(i) - dec; }; return res;").param("dec", 1)
+                        .percentiles(pcts))
+                .execute().actionGet();
+
+        assertThat(searchResponse.getHits().getTotalHits(), equalTo(10l));
+
+        final PercentileRanks percentiles = searchResponse.getAggregations().get("percentile_ranks");
+        assertConsistent(pcts, percentiles, minValues - 1, maxValues - 1);
+    }
+
+    @Test
+    public void testOrderBySubAggregation() {
+        boolean asc = randomBoolean();
+        SearchResponse searchResponse = client().prepareSearch("idx")
+                .setQuery(matchAllQuery())
+                .addAggregation(
+                        histogram("histo").field("value").interval(2l)
+                            .subAggregation(randomCompression(percentileRanks("percentile_ranks").percentiles(99)))
+                            .order(Order.aggregation("percentile_ranks", "99", asc)))
+                .execute().actionGet();
+
+        assertThat(searchResponse.getHits().getTotalHits(), equalTo(10l));
+
+        Histogram histo = searchResponse.getAggregations().get("histo");
+        double previous = asc ? Double.NEGATIVE_INFINITY : Double.POSITIVE_INFINITY;
+        for (Histogram.Bucket bucket : histo.getBuckets()) {
+            PercentileRanks percentiles = bucket.getAggregations().get("percentile_ranks");
+            double p99 = percentiles.percent(99);
+            if (asc) {
+                assertThat(p99, greaterThanOrEqualTo(previous));
+            } else {
+                assertThat(p99, lessThanOrEqualTo(previous));
+            }
+            previous = p99;
+        }
+    }
+
+}

src/test/java/org/elasticsearch/search/aggregations/metrics/PercentilesTests.java

@@ -23,8 +23,8 @@ import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.common.logging.Loggers;
 import org.elasticsearch.search.aggregations.bucket.histogram.Histogram;
 import org.elasticsearch.search.aggregations.bucket.histogram.Histogram.Order;
+import org.elasticsearch.search.aggregations.metrics.percentiles.Percentile;
 import org.elasticsearch.search.aggregations.metrics.percentiles.Percentiles;
-import org.elasticsearch.search.aggregations.metrics.percentiles.Percentiles.Percentile;
 import org.elasticsearch.search.aggregations.metrics.percentiles.PercentilesBuilder;
 import org.junit.Test;