Supporting sort modes avg and sum when sorting inside nested objects.

Previously this commit either sort modes `min` or `max` (depending on sort order) was used when sort modes `avg` and `sum` were picked.

Closes #2701

src/main/java/org/elasticsearch/index/fielddata/fieldcomparator/ByteValuesComparator.java

@@ -56,4 +56,14 @@ public final class ByteValuesComparator extends LongValuesComparatorBase<Byte> {
     public Byte value(int slot) {
         return Byte.valueOf(values[slot]);
     }
+
+    @Override
+    public void add(int slot, int doc) {
+        values[slot] += (byte) readerValues.getValueMissing(doc, missingValue);
+    }
+
+    @Override
+    public void divide(int slot, int divisor) {
+        values[slot] /= divisor;
+    }
 }

src/main/java/org/elasticsearch/index/fielddata/fieldcomparator/DoubleValuesComparator.java

@@ -56,4 +56,14 @@ public final class DoubleValuesComparator extends DoubleValuesComparatorBase<Dou
     public Double value(int slot) {
         return Double.valueOf(values[slot]);
     }
+
+    @Override
+    public void add(int slot, int doc) {
+        values[slot] += readerValues.getValueMissing(doc, missingValue);
+    }
+
+    @Override
+    public void divide(int slot, int divisor) {
+        values[slot] /= divisor;
+    }
 }

src/main/java/org/elasticsearch/index/fielddata/fieldcomparator/DoubleValuesComparatorBase.java

@@ -25,7 +25,7 @@ import org.elasticsearch.index.fielddata.IndexNumericFieldData;
 
 import java.io.IOException;
 
-abstract class DoubleValuesComparatorBase<T extends Number> extends FieldComparator<T> {
+abstract class DoubleValuesComparatorBase<T extends Number> extends NumberComparatorBase<T> {
 
     protected final IndexNumericFieldData<?> indexFieldData;
     protected final double missingValue;

src/main/java/org/elasticsearch/index/fielddata/fieldcomparator/FloatValuesComparator.java

@@ -63,4 +63,13 @@ public final class FloatValuesComparator extends DoubleValuesComparatorBase<Floa
         return Float.valueOf(values[slot]);
     }
 
+    @Override
+    public void add(int slot, int doc) {
+        values[slot] += (float) readerValues.getValueMissing(doc, missingValue);
+    }
+
+    @Override
+    public void divide(int slot, int divisor) {
+        values[slot] /= divisor;
+    }
 }

src/main/java/org/elasticsearch/index/fielddata/fieldcomparator/IntValuesComparator.java

@@ -62,4 +62,14 @@ public final class IntValuesComparator extends LongValuesComparatorBase<Integer>
     public Integer value(int slot) {
         return Integer.valueOf(values[slot]);
     }
+
+    @Override
+    public void add(int slot, int doc) {
+        values[slot] += (int) readerValues.getValueMissing(doc, missingValue);
+    }
+
+    @Override
+    public void divide(int slot, int divisor) {
+        values[slot] /= divisor;
+    }
 }

src/main/java/org/elasticsearch/index/fielddata/fieldcomparator/LongValuesComparator.java

@@ -55,4 +55,14 @@ public final class LongValuesComparator extends LongValuesComparatorBase<Long> {
     public Long value(int slot) {
         return Long.valueOf(values[slot]);
     }
+
+    @Override
+    public void add(int slot, int doc) {
+        values[slot] += readerValues.getValueMissing(doc, missingValue);
+    }
+
+    @Override
+    public void divide(int slot, int divisor) {
+        values[slot] /= divisor;
+    }
 }

src/main/java/org/elasticsearch/index/fielddata/fieldcomparator/LongValuesComparatorBase.java

@@ -25,7 +25,7 @@ import org.elasticsearch.index.fielddata.LongValues;
 
 import java.io.IOException;
 
-abstract class LongValuesComparatorBase<T extends Number> extends FieldComparator<T> {
+abstract class LongValuesComparatorBase<T extends Number> extends NumberComparatorBase<T> {
 
     protected final IndexNumericFieldData<?> indexFieldData;
     protected final long missingValue;

src/main/java/org/elasticsearch/index/fielddata/fieldcomparator/NumberComparatorBase.java

@@ -0,0 +1,45 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon 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.index.fielddata.fieldcomparator;
+
+import org.apache.lucene.search.FieldComparator;
+
+/**
+ * Base FieldComparator class for number fields.
+ */
+// This is right now only used for sorting number based fields inside nested objects
+public abstract class NumberComparatorBase<T> extends FieldComparator<T> {
+
+    /**
+     * Adds numeric value at the specified doc to the specified slot.
+     *
+     * @param slot  The specified slot
+     * @param doc   The specified doc
+     */
+    public abstract void add(int slot, int doc);
+
+    /**
+     * Divides the value at the specified slot with the specified divisor.
+     *
+     * @param slot      The specified slot
+     * @param divisor   The specified divisor
+     */
+    public abstract void divide(int slot, int divisor);
+}

src/main/java/org/elasticsearch/index/fielddata/fieldcomparator/ShortValuesComparator.java

@@ -59,4 +59,13 @@ public final class ShortValuesComparator extends LongValuesComparatorBase<Short>
         return Short.valueOf(values[slot]);
     }
 
+    @Override
+    public void add(int slot, int doc) {
+        values[slot] += (short) readerValues.getValueMissing(doc, missingValue);
+    }
+
+    @Override
+    public void divide(int slot, int divisor) {
+        values[slot] /= divisor;
+    }
 }

src/main/java/org/elasticsearch/index/search/nested/NestedFieldComparatorSource.java

@@ -1,3 +1,22 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon 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.index.search.nested;
 
 import org.apache.lucene.index.AtomicReaderContext;
@@ -9,6 +28,7 @@ import org.apache.lucene.util.FixedBitSet;
 import org.elasticsearch.ElasticSearchIllegalArgumentException;
 import org.elasticsearch.common.lucene.docset.DocIdSets;
 import org.elasticsearch.index.fielddata.IndexFieldData;
+import org.elasticsearch.index.fielddata.fieldcomparator.NumberComparatorBase;
 import org.elasticsearch.index.fielddata.fieldcomparator.SortMode;
 
 import java.io.IOException;
@@ -38,6 +58,10 @@ public class NestedFieldComparatorSource extends IndexFieldData.XFieldComparator
                 return new NestedFieldComparator.Highest(wrappedComparator, rootDocumentsFilter, innerDocumentsFilter, numHits);
             case MIN:
                 return new NestedFieldComparator.Lowest(wrappedComparator, rootDocumentsFilter, innerDocumentsFilter, numHits);
+            case SUM:
+                return new Sum((NumberComparatorBase) wrappedComparator, rootDocumentsFilter, innerDocumentsFilter, numHits);
+            case AVG:
+                return new Avg((NumberComparatorBase) wrappedComparator, rootDocumentsFilter, innerDocumentsFilter, numHits);
             default:
                 throw new ElasticSearchIllegalArgumentException(
                         String.format("Unsupported sort_mode[%s] for nested type", sortMode)
@@ -51,6 +75,165 @@ public class NestedFieldComparatorSource extends IndexFieldData.XFieldComparator
     }
 }
 
+class Sum extends FieldComparator {
+
+    final Filter rootDocumentsFilter;
+    final Filter innerDocumentsFilter;
+    final int spareSlot;
+
+    NumberComparatorBase wrappedComparator;
+    FixedBitSet rootDocuments;
+    FixedBitSet innerDocuments;
+    int bottomSlot;
+
+    Sum(NumberComparatorBase wrappedComparator, Filter rootDocumentsFilter, Filter innerDocumentsFilter, int spareSlot) {
+        this.wrappedComparator = wrappedComparator;
+        this.rootDocumentsFilter = rootDocumentsFilter;
+        this.innerDocumentsFilter = innerDocumentsFilter;
+        this.spareSlot = spareSlot;
+    }
+
+    @Override
+    public int compare(int slot1, int slot2) {
+        return wrappedComparator.compare(slot1, slot2);
+    }
+
+    @Override
+    public void setBottom(int slot) {
+        wrappedComparator.setBottom(slot);
+        this.bottomSlot = slot;
+    }
+
+    @Override
+    public FieldComparator setNextReader(AtomicReaderContext context) throws IOException {
+        DocIdSet innerDocuments = innerDocumentsFilter.getDocIdSet(context, null);
+        if (DocIdSets.isEmpty(innerDocuments)) {
+            this.innerDocuments = null;
+        } else if (innerDocuments instanceof FixedBitSet) {
+            this.innerDocuments = (FixedBitSet) innerDocuments;
+        } else {
+            this.innerDocuments = DocIdSets.toFixedBitSet(innerDocuments.iterator(), context.reader().maxDoc());
+        }
+        DocIdSet rootDocuments = rootDocumentsFilter.getDocIdSet(context, null);
+        if (DocIdSets.isEmpty(rootDocuments)) {
+            this.rootDocuments = null;
+        } else if (rootDocuments instanceof FixedBitSet) {
+            this.rootDocuments = (FixedBitSet) rootDocuments;
+        } else {
+            this.rootDocuments = DocIdSets.toFixedBitSet(rootDocuments.iterator(), context.reader().maxDoc());
+        }
+
+        wrappedComparator = (NumberComparatorBase) wrappedComparator.setNextReader(context);
+        return this;
+    }
+
+    @Override
+    public Object value(int slot) {
+        return wrappedComparator.value(slot);
+    }
+
+    @Override
+    public int compareBottom(int rootDoc) throws IOException {
+        if (rootDoc == 0 || rootDocuments == null || innerDocuments == null) {
+            return 0;
+        }
+
+        int prevRootDoc = rootDocuments.prevSetBit(rootDoc - 1);
+        int nestedDoc = innerDocuments.nextSetBit(prevRootDoc + 1);
+        if (nestedDoc >= rootDoc || nestedDoc == -1) {
+            return 0;
+        }
+
+        wrappedComparator.copy(spareSlot, nestedDoc);
+        nestedDoc = innerDocuments.nextSetBit(nestedDoc + 1);
+        while (nestedDoc > prevRootDoc && nestedDoc < rootDoc) {
+            wrappedComparator.add(spareSlot, nestedDoc);
+            nestedDoc = innerDocuments.nextSetBit(nestedDoc + 1);
+        }
+        return compare(bottomSlot, spareSlot);
+    }
+
+    @Override
+    public void copy(int slot, int rootDoc) throws IOException {
+        if (rootDoc == 0 || rootDocuments == null || innerDocuments == null) {
+            return;
+        }
+
+        int prevRootDoc = rootDocuments.prevSetBit(rootDoc - 1);
+        int nestedDoc = innerDocuments.nextSetBit(prevRootDoc + 1);
+        if (nestedDoc >= rootDoc || nestedDoc == -1) {
+            return;
+        }
+
+        wrappedComparator.copy(slot, nestedDoc);
+        nestedDoc = innerDocuments.nextSetBit(nestedDoc + 1);
+        while (nestedDoc > prevRootDoc && nestedDoc < rootDoc) {
+            wrappedComparator.add(slot, nestedDoc);
+            nestedDoc = innerDocuments.nextSetBit(nestedDoc + 1);
+        }
+    }
+
+    @Override
+    public int compareDocToValue(int rootDoc, Object value) throws IOException {
+        throw new UnsupportedOperationException("compareDocToValue() not used for sorting in ES");
+    }
+}
+
+final class Avg extends Sum {
+
+    Avg(NumberComparatorBase wrappedComparator, Filter rootDocumentsFilter, Filter innerDocumentsFilter, int spareSlot) {
+        super(wrappedComparator, rootDocumentsFilter, innerDocumentsFilter, spareSlot);
+    }
+
+    @Override
+    public int compareBottom(int rootDoc) throws IOException {
+        if (rootDoc == 0 || rootDocuments == null || innerDocuments == null) {
+            return 0;
+        }
+
+        int prevRootDoc = rootDocuments.prevSetBit(rootDoc - 1);
+        int nestedDoc = innerDocuments.nextSetBit(prevRootDoc + 1);
+        if (nestedDoc >= rootDoc || nestedDoc == -1) {
+            return 0;
+        }
+
+        int counter = 1;
+        wrappedComparator.copy(spareSlot, nestedDoc);
+        nestedDoc = innerDocuments.nextSetBit(nestedDoc + 1);
+        while (nestedDoc > prevRootDoc && nestedDoc < rootDoc) {
+            wrappedComparator.add(spareSlot, nestedDoc);
+            nestedDoc = innerDocuments.nextSetBit(nestedDoc + 1);
+            counter++;
+        }
+        wrappedComparator.divide(spareSlot, counter);
+        return compare(bottomSlot, spareSlot);
+    }
+
+    @Override
+    public void copy(int slot, int rootDoc) throws IOException {
+        if (rootDoc == 0 || rootDocuments == null || innerDocuments == null) {
+            return;
+        }
+
+        int prevRootDoc = rootDocuments.prevSetBit(rootDoc - 1);
+        int nestedDoc = innerDocuments.nextSetBit(prevRootDoc + 1);
+        if (nestedDoc >= rootDoc || nestedDoc == -1) {
+            return;
+        }
+
+        int counter = 1;
+        wrappedComparator.copy(slot, nestedDoc);
+        nestedDoc = innerDocuments.nextSetBit(nestedDoc + 1);
+        while (nestedDoc > prevRootDoc && nestedDoc < rootDoc) {
+            wrappedComparator.add(slot, nestedDoc);
+            nestedDoc = innerDocuments.nextSetBit(nestedDoc + 1);
+            counter++;
+        }
+        wrappedComparator.divide(slot, counter);
+    }
+}
+
+// Move to Lucene join module
 abstract class NestedFieldComparator extends FieldComparator {
 
     final Filter rootDocumentsFilter;

src/main/java/org/elasticsearch/search/sort/SortParseElement.java

@@ -224,10 +224,6 @@ public class SortParseElement implements SearchParseElement {
                 } else {
                     innerDocumentsFilter = context.filterCache().cache(objectMapper.nestedTypeFilter());
                 }
-                // For now the nested sorting doesn't support SUM or AVG
-                if (sortMode == SortMode.SUM || sortMode == SortMode.AVG) {
-                    sortMode = resolveDefaultSortMode(reverse);
-                }
                 fieldComparatorSource = new NestedFieldComparatorSource(sortMode, fieldComparatorSource, rootDocumentsFilter, innerDocumentsFilter);
             }
             sortFields.add(new SortField(fieldMapper.names().indexName(), fieldComparatorSource, reverse));

src/test/java/org/elasticsearch/test/integration/nested/SimpleNestedTests.java

@@ -706,6 +706,7 @@ public class SimpleNestedTests extends AbstractNodesTests {
                 .execute().actionGet();
         client.admin().cluster().prepareHealth().setWaitForGreenStatus().execute().actionGet();
 
+        // sum: 11
         client.prepareIndex("test", "type1", Integer.toString(1)).setSource(jsonBuilder().startObject()
                 .field("grand_parent_values", 1l)
                 .startObject("parent")
@@ -737,6 +738,7 @@ public class SimpleNestedTests extends AbstractNodesTests {
                 .endObject()
                 .endObject()).execute().actionGet();
 
+        // sum: 7
         client.prepareIndex("test", "type1", Integer.toString(2)).setSource(jsonBuilder().startObject()
                 .field("grand_parent_values", 2l)
                 .startObject("parent")
@@ -768,6 +770,7 @@ public class SimpleNestedTests extends AbstractNodesTests {
                 .endObject()
                 .endObject()).execute().actionGet();
 
+        // sum: 2
         client.prepareIndex("test", "type1", Integer.toString(3)).setSource(jsonBuilder().startObject()
                 .field("grand_parent_values", 3l)
                 .startObject("parent")
@@ -911,6 +914,121 @@ public class SimpleNestedTests extends AbstractNodesTests {
         assertThat(searchResponse.getHits().getHits()[1].sortValues()[0].toString(), equalTo("2"));
         assertThat(searchResponse.getHits().getHits()[2].getId(), equalTo("3"));
         assertThat(searchResponse.getHits().getHits()[2].sortValues()[0].toString(), equalTo("3"));
+
+        // Sort mode: sum
+        searchResponse = client.prepareSearch()
+                .setQuery(matchAllQuery())
+                .addSort(
+                        SortBuilders.fieldSort("parent.child.child_values")
+                                .setNestedPath("parent.child")
+                                .sortMode("sum")
+                                .order(SortOrder.ASC)
+                )
+                .execute().actionGet();
+        assertThat(searchResponse.getHits().totalHits(), equalTo(3l));
+        assertThat(searchResponse.getHits().getHits().length, equalTo(3));
+        assertThat(searchResponse.getHits().getHits()[0].getId(), equalTo("3"));
+        assertThat(searchResponse.getHits().getHits()[0].sortValues()[0].toString(), equalTo("2"));
+        assertThat(searchResponse.getHits().getHits()[1].getId(), equalTo("2"));
+        assertThat(searchResponse.getHits().getHits()[1].sortValues()[0].toString(), equalTo("7"));
+        assertThat(searchResponse.getHits().getHits()[2].getId(), equalTo("1"));
+        assertThat(searchResponse.getHits().getHits()[2].sortValues()[0].toString(), equalTo("11"));
+
+
+        searchResponse = client.prepareSearch()
+                .setQuery(matchAllQuery())
+                .addSort(
+                        SortBuilders.fieldSort("parent.child.child_values")
+                                .setNestedPath("parent.child")
+                                .sortMode("sum")
+                                .order(SortOrder.DESC)
+                )
+                .execute().actionGet();
+        assertThat(searchResponse.getHits().totalHits(), equalTo(3l));
+        assertThat(searchResponse.getHits().getHits().length, equalTo(3));
+        assertThat(searchResponse.getHits().getHits()[0].getId(), equalTo("1"));
+        assertThat(searchResponse.getHits().getHits()[0].sortValues()[0].toString(), equalTo("11"));
+        assertThat(searchResponse.getHits().getHits()[1].getId(), equalTo("2"));
+        assertThat(searchResponse.getHits().getHits()[1].sortValues()[0].toString(), equalTo("7"));
+        assertThat(searchResponse.getHits().getHits()[2].getId(), equalTo("3"));
+        assertThat(searchResponse.getHits().getHits()[2].sortValues()[0].toString(), equalTo("2"));
+
+        // Sort mode: sum with filter
+        searchResponse = client.prepareSearch()
+                .setQuery(matchAllQuery())
+                .addSort(
+                        SortBuilders.fieldSort("parent.child.child_values")
+                                .setNestedPath("parent.child")
+                                .setNestedFilter(FilterBuilders.termFilter("parent.child.filter", true))
+                                .sortMode("sum")
+                                .order(SortOrder.ASC)
+                )
+                .execute().actionGet();
+        assertThat(searchResponse.getHits().totalHits(), equalTo(3l));
+        assertThat(searchResponse.getHits().getHits().length, equalTo(3));
+        assertThat(searchResponse.getHits().getHits()[0].getId(), equalTo("1"));
+        assertThat(searchResponse.getHits().getHits()[0].sortValues()[0].toString(), equalTo("1"));
+        assertThat(searchResponse.getHits().getHits()[1].getId(), equalTo("2"));
+        assertThat(searchResponse.getHits().getHits()[1].sortValues()[0].toString(), equalTo("2"));
+        assertThat(searchResponse.getHits().getHits()[2].getId(), equalTo("3"));
+        assertThat(searchResponse.getHits().getHits()[2].sortValues()[0].toString(), equalTo("3"));
+
+        // Sort mode: avg
+        searchResponse = client.prepareSearch()
+                .setQuery(matchAllQuery())
+                .addSort(
+                        SortBuilders.fieldSort("parent.child.child_values")
+                                .setNestedPath("parent.child")
+                                .sortMode("avg")
+                                .order(SortOrder.ASC)
+                )
+                .execute().actionGet();
+        assertThat(searchResponse.getHits().totalHits(), equalTo(3l));
+        assertThat(searchResponse.getHits().getHits().length, equalTo(3));
+        assertThat(searchResponse.getHits().getHits()[0].getId(), equalTo("3"));
+        assertThat(searchResponse.getHits().getHits()[0].sortValues()[0].toString(), equalTo("0"));
+        assertThat(searchResponse.getHits().getHits()[1].getId(), equalTo("2"));
+        assertThat(searchResponse.getHits().getHits()[1].sortValues()[0].toString(), equalTo("1"));
+        assertThat(searchResponse.getHits().getHits()[2].getId(), equalTo("1"));
+        assertThat(searchResponse.getHits().getHits()[2].sortValues()[0].toString(), equalTo("2"));
+
+        searchResponse = client.prepareSearch()
+                .setQuery(matchAllQuery())
+                .addSort(
+                        SortBuilders.fieldSort("parent.child.child_values")
+                                .setNestedPath("parent.child")
+                                .sortMode("avg")
+                                .order(SortOrder.DESC)
+                )
+                .execute().actionGet();
+        assertThat(searchResponse.getHits().totalHits(), equalTo(3l));
+        assertThat(searchResponse.getHits().getHits().length, equalTo(3));
+        assertThat(searchResponse.getHits().getHits()[0].getId(), equalTo("1"));
+        assertThat(searchResponse.getHits().getHits()[0].sortValues()[0].toString(), equalTo("2"));
+        assertThat(searchResponse.getHits().getHits()[1].getId(), equalTo("2"));
+        assertThat(searchResponse.getHits().getHits()[1].sortValues()[0].toString(), equalTo("1"));
+        assertThat(searchResponse.getHits().getHits()[2].getId(), equalTo("3"));
+        assertThat(searchResponse.getHits().getHits()[2].sortValues()[0].toString(), equalTo("0"));
+
+        // Sort mode: avg with filter
+        searchResponse = client.prepareSearch()
+                .setQuery(matchAllQuery())
+                .addSort(
+                        SortBuilders.fieldSort("parent.child.child_values")
+                                .setNestedPath("parent.child")
+                                .setNestedFilter(FilterBuilders.termFilter("parent.child.filter", true))
+                                .sortMode("avg")
+                                .order(SortOrder.ASC)
+                )
+                .execute().actionGet();
+        assertThat(searchResponse.getHits().totalHits(), equalTo(3l));
+        assertThat(searchResponse.getHits().getHits().length, equalTo(3));
+        assertThat(searchResponse.getHits().getHits()[0].getId(), equalTo("1"));
+        assertThat(searchResponse.getHits().getHits()[0].sortValues()[0].toString(), equalTo("1"));
+        assertThat(searchResponse.getHits().getHits()[1].getId(), equalTo("2"));
+        assertThat(searchResponse.getHits().getHits()[1].sortValues()[0].toString(), equalTo("2"));
+        assertThat(searchResponse.getHits().getHits()[2].getId(), equalTo("3"));
+        assertThat(searchResponse.getHits().getHits()[2].sortValues()[0].toString(), equalTo("3"));
     }
 
 }

src/test/java/org/elasticsearch/test/unit/index/search/nested/AbstractNumberNestedSortingTests.java

@@ -0,0 +1,314 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon 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.test.unit.index.search.nested;
+
+import org.apache.lucene.document.Document;
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.StringField;
+import org.apache.lucene.index.DirectoryReader;
+import org.apache.lucene.index.IndexableField;
+import org.apache.lucene.index.Term;
+import org.apache.lucene.search.*;
+import org.apache.lucene.search.join.ScoreMode;
+import org.apache.lucene.search.join.ToParentBlockJoinQuery;
+import org.elasticsearch.common.lucene.search.AndFilter;
+import org.elasticsearch.common.lucene.search.NotFilter;
+import org.elasticsearch.common.lucene.search.TermFilter;
+import org.elasticsearch.common.lucene.search.XFilteredQuery;
+import org.elasticsearch.index.fielddata.IndexFieldData;
+import org.elasticsearch.index.fielddata.fieldcomparator.SortMode;
+import org.elasticsearch.index.search.nested.NestedFieldComparatorSource;
+import org.elasticsearch.test.unit.index.fielddata.AbstractFieldDataTests;
+import org.testng.annotations.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+
+/**
+ */
+public abstract class AbstractNumberNestedSortingTests extends AbstractFieldDataTests {
+
+    @Test
+    public void testNestedSorting() throws Exception {
+        List<Document> docs = new ArrayList<Document>();
+        Document document = new Document();
+        document.add(createField("field2", 3, Field.Store.NO));
+        document.add(new StringField("filter_1", "T", Field.Store.NO));
+        docs.add(document);
+        document = new Document();
+        document.add(createField("field2", 3, Field.Store.NO));
+        document.add(new StringField("filter_1", "T", Field.Store.NO));
+        docs.add(document);
+        document = new Document();
+        document.add(createField("field2", 3, Field.Store.NO));
+        document.add(new StringField("filter_1", "T", Field.Store.NO));
+        docs.add(document);
+        document = new Document();
+        document.add(new StringField("__type", "parent", Field.Store.NO));
+        document.add(createField("field1", 1, Field.Store.NO));
+        docs.add(document);
+        writer.addDocuments(docs);
+        writer.commit();
+
+        docs.clear();
+        document = new Document();
+        document.add(createField("field2", 3, Field.Store.NO));
+        document.add(new StringField("filter_1", "T", Field.Store.NO));
+        docs.add(document);
+        document = new Document();
+        document.add(createField("field2", 3, Field.Store.NO));
+        document.add(new StringField("filter_1", "T", Field.Store.NO));
+        docs.add(document);
+        document = new Document();
+        document.add(createField("field2", 2, Field.Store.NO));
+        document.add(new StringField("filter_1", "T", Field.Store.NO));
+        docs.add(document);
+        document = new Document();
+        document.add(new StringField("__type", "parent", Field.Store.NO));
+        document.add(createField("field1", 2, Field.Store.NO));
+        docs.add(document);
+        writer.addDocuments(docs);
+
+        docs.clear();
+        document = new Document();
+        document.add(createField("field2", 3, Field.Store.NO));
+        document.add(new StringField("filter_1", "T", Field.Store.NO));
+        docs.add(document);
+        document = new Document();
+        document.add(createField("field2", 3, Field.Store.NO));
+        document.add(new StringField("filter_1", "T", Field.Store.NO));
+        docs.add(document);
+        document = new Document();
+        document.add(createField("field2", 1, Field.Store.NO));
+        document.add(new StringField("filter_1", "T", Field.Store.NO));
+        docs.add(document);
+        document = new Document();
+        document.add(new StringField("__type", "parent", Field.Store.NO));
+        document.add(createField("field1", 3, Field.Store.NO));
+        docs.add(document);
+        writer.addDocuments(docs);
+
+        docs.clear();
+        document = new Document();
+        document.add(createField("field2", 3, Field.Store.NO));
+        document.add(new StringField("filter_1", "T", Field.Store.NO));
+        docs.add(document);
+        document = new Document();
+        document.add(createField("field2", 3, Field.Store.NO));
+        document.add(new StringField("filter_1", "F", Field.Store.NO));
+        docs.add(document);
+        document = new Document();
+        document.add(createField("field2", 4, Field.Store.NO));
+        document.add(new StringField("filter_1", "F", Field.Store.NO));
+        docs.add(document);
+        document = new Document();
+        document.add(new StringField("__type", "parent", Field.Store.NO));
+        document.add(createField("field1", 4, Field.Store.NO));
+        docs.add(document);
+        writer.addDocuments(docs);
+        writer.commit();
+
+        docs.clear();
+        document = new Document();
+        document.add(createField("field2", 3, Field.Store.NO));
+        document.add(new StringField("filter_1", "F", Field.Store.NO));
+        docs.add(document);
+        document = new Document();
+        document.add(createField("field2", 3, Field.Store.NO));
+        document.add(new StringField("filter_1", "F", Field.Store.NO));
+        docs.add(document);
+        document = new Document();
+        document.add(createField("field2", 5, Field.Store.NO));
+        document.add(new StringField("filter_1", "F", Field.Store.NO));
+        docs.add(document);
+        document = new Document();
+        document.add(new StringField("__type", "parent", Field.Store.NO));
+        document.add(createField("field1", 5, Field.Store.NO));
+        docs.add(document);
+        writer.addDocuments(docs);
+
+        docs.clear();
+        document = new Document();
+        document.add(createField("field2", 3, Field.Store.NO));
+        document.add(new StringField("filter_1", "T", Field.Store.NO));
+        docs.add(document);
+        document = new Document();
+        document.add(createField("field2", 3, Field.Store.NO));
+        document.add(new StringField("filter_1", "T", Field.Store.NO));
+        docs.add(document);
+        document = new Document();
+        document.add(createField("field2", 6, Field.Store.NO));
+        document.add(new StringField("filter_1", "T", Field.Store.NO));
+        docs.add(document);
+        document = new Document();
+        document.add(new StringField("__type", "parent", Field.Store.NO));
+        document.add(createField("field1", 6, Field.Store.NO));
+        docs.add(document);
+        writer.addDocuments(docs);
+
+        // This doc will not be included, because it doesn't have nested docs
+        document = new Document();
+        document.add(new StringField("__type", "parent", Field.Store.NO));
+        document.add(createField("field1", 7, Field.Store.NO));
+        writer.addDocument(document);
+
+        docs.clear();
+        document = new Document();
+        document.add(createField("field2", 3, Field.Store.NO));
+        document.add(new StringField("filter_1", "T", Field.Store.NO));
+        docs.add(document);
+        document = new Document();
+        document.add(createField("field2", 3, Field.Store.NO));
+        document.add(new StringField("filter_1", "F", Field.Store.NO));
+        docs.add(document);
+        document = new Document();
+        document.add(createField("field2", 7, Field.Store.NO));
+        document.add(new StringField("filter_1", "F", Field.Store.NO));
+        docs.add(document);
+        document = new Document();
+        document.add(new StringField("__type", "parent", Field.Store.NO));
+        document.add(createField("field1", 8, Field.Store.NO));
+        docs.add(document);
+        writer.addDocuments(docs);
+        writer.commit();
+
+        // Some garbage docs, just to check if the NestedFieldComparator can deal with this.
+        document = new Document();
+        document.add(new StringField("fieldXXX", "x", Field.Store.NO));
+        writer.addDocument(document);
+        document = new Document();
+        document.add(new StringField("fieldXXX", "x", Field.Store.NO));
+        writer.addDocument(document);
+        document = new Document();
+        document.add(new StringField("fieldXXX", "x", Field.Store.NO));
+        writer.addDocument(document);
+
+        SortMode sortMode = SortMode.SUM;
+        IndexSearcher searcher = new IndexSearcher(DirectoryReader.open(writer, false));
+        IndexFieldData.XFieldComparatorSource innerFieldComparator = createInnerFieldComparator("field2", sortMode);
+        Filter parentFilter = new TermFilter(new Term("__type", "parent"));
+        Filter childFilter = new NotFilter(parentFilter);
+        NestedFieldComparatorSource nestedComparatorSource = new NestedFieldComparatorSource(sortMode, innerFieldComparator, parentFilter, childFilter);
+        ToParentBlockJoinQuery query = new ToParentBlockJoinQuery(new XFilteredQuery(new MatchAllDocsQuery(), childFilter), new CachingWrapperFilter(parentFilter), ScoreMode.None);
+
+        Sort sort = new Sort(new SortField("field2", nestedComparatorSource));
+        TopFieldDocs topDocs = searcher.search(query, 5, sort);
+        assertThat(topDocs.totalHits, equalTo(7));
+        assertThat(topDocs.scoreDocs.length, equalTo(5));
+        assertThat(topDocs.scoreDocs[0].doc, equalTo(11));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[0]).fields[0]).intValue(), equalTo(7));
+        assertThat(topDocs.scoreDocs[1].doc, equalTo(7));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[1]).fields[0]).intValue(), equalTo(8));
+        assertThat(topDocs.scoreDocs[2].doc, equalTo(3));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[2]).fields[0]).intValue(), equalTo(9));
+        assertThat(topDocs.scoreDocs[3].doc, equalTo(15));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[3]).fields[0]).intValue(), equalTo(10));
+        assertThat(topDocs.scoreDocs[4].doc, equalTo(19));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[4]).fields[0]).intValue(), equalTo(11));
+
+        sort = new Sort(new SortField("field2", nestedComparatorSource, true));
+        topDocs = searcher.search(query, 5, sort);
+        assertThat(topDocs.totalHits, equalTo(7));
+        assertThat(topDocs.scoreDocs.length, equalTo(5));
+        assertThat(topDocs.scoreDocs[0].doc, equalTo(28));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[0]).fields[0]).intValue(), equalTo(13));
+        assertThat(topDocs.scoreDocs[1].doc, equalTo(23));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[1]).fields[0]).intValue(), equalTo(12));
+        assertThat(topDocs.scoreDocs[2].doc, equalTo(19));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[2]).fields[0]).intValue(), equalTo(11));
+        assertThat(topDocs.scoreDocs[3].doc, equalTo(15));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[3]).fields[0]).intValue(), equalTo(10));
+        assertThat(topDocs.scoreDocs[4].doc, equalTo(3));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[4]).fields[0]).intValue(), equalTo(9));
+
+        childFilter = new AndFilter(Arrays.asList(new NotFilter(parentFilter), new TermFilter(new Term("filter_1", "T"))));
+        nestedComparatorSource = new NestedFieldComparatorSource(sortMode, innerFieldComparator, parentFilter, childFilter);
+        query = new ToParentBlockJoinQuery(
+                new XFilteredQuery(new MatchAllDocsQuery(), childFilter),
+                new CachingWrapperFilter(parentFilter),
+                ScoreMode.None
+        );
+        sort = new Sort(new SortField("field2", nestedComparatorSource, true));
+        topDocs = searcher.search(query, 5, sort);
+        assertThat(topDocs.totalHits, equalTo(6));
+        assertThat(topDocs.scoreDocs.length, equalTo(5));
+        assertThat(topDocs.scoreDocs[0].doc, equalTo(23));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[0]).fields[0]).intValue(), equalTo(12));
+        assertThat(topDocs.scoreDocs[1].doc, equalTo(3));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[1]).fields[0]).intValue(), equalTo(9));
+        assertThat(topDocs.scoreDocs[2].doc, equalTo(7));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[2]).fields[0]).intValue(), equalTo(8));
+        assertThat(topDocs.scoreDocs[3].doc, equalTo(11));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[3]).fields[0]).intValue(), equalTo(7));
+        assertThat(topDocs.scoreDocs[4].doc, equalTo(15));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[4]).fields[0]).intValue(), equalTo(3));
+
+        sort = new Sort(new SortField("field2", nestedComparatorSource));
+        topDocs = searcher.search(query, 5, sort);
+        assertThat(topDocs.totalHits, equalTo(6));
+        assertThat(topDocs.scoreDocs.length, equalTo(5));
+        assertThat(topDocs.scoreDocs[0].doc, equalTo(15));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[0]).fields[0]).intValue(), equalTo(3));
+        assertThat(topDocs.scoreDocs[1].doc, equalTo(28));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[1]).fields[0]).intValue(), equalTo(3));
+        assertThat(topDocs.scoreDocs[2].doc, equalTo(11));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[2]).fields[0]).intValue(), equalTo(7));
+        assertThat(topDocs.scoreDocs[3].doc, equalTo(7));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[3]).fields[0]).intValue(), equalTo(8));
+        assertThat(topDocs.scoreDocs[4].doc, equalTo(3));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[4]).fields[0]).intValue(), equalTo(9));
+
+        // Moved to method, because floating point based XFieldComparatorSource have different outcome for SortMode avg,
+        // than integral number based implementations...
+        assertAvgScoreMode(parentFilter, searcher, innerFieldComparator);
+        searcher.getIndexReader().close();
+    }
+
+    protected void assertAvgScoreMode(Filter parentFilter, IndexSearcher searcher, IndexFieldData.XFieldComparatorSource innerFieldComparator) throws IOException {
+        SortMode sortMode = SortMode.AVG;
+        Filter childFilter = new NotFilter(parentFilter);
+        NestedFieldComparatorSource nestedComparatorSource = new NestedFieldComparatorSource(sortMode, innerFieldComparator, parentFilter, childFilter);
+        Query query = new ToParentBlockJoinQuery(new XFilteredQuery(new MatchAllDocsQuery(), childFilter), new CachingWrapperFilter(parentFilter), ScoreMode.None);
+        Sort sort = new Sort(new SortField("field2", nestedComparatorSource));
+        TopDocs topDocs = searcher.search(query, 5, sort);
+        assertThat(topDocs.totalHits, equalTo(7));
+        assertThat(topDocs.scoreDocs.length, equalTo(5));
+        assertThat(topDocs.scoreDocs[0].doc, equalTo(7));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[0]).fields[0]).intValue(), equalTo(2));
+        assertThat(topDocs.scoreDocs[1].doc, equalTo(11));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[1]).fields[0]).intValue(), equalTo(2));
+        assertThat(topDocs.scoreDocs[2].doc, equalTo(3));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[2]).fields[0]).intValue(), equalTo(3));
+        assertThat(topDocs.scoreDocs[3].doc, equalTo(15));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[3]).fields[0]).intValue(), equalTo(3));
+        assertThat(topDocs.scoreDocs[4].doc, equalTo(19));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[4]).fields[0]).intValue(), equalTo(3));
+    }
+
+    protected abstract IndexableField createField(String name, int value, Field.Store store);
+
+    protected abstract IndexFieldData.XFieldComparatorSource createInnerFieldComparator(String fieldName, SortMode sortMode);
+
+}

src/test/java/org/elasticsearch/test/unit/index/search/nested/ByteNestedSortingTests.java

@@ -0,0 +1,50 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon 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.test.unit.index.search.nested;
+
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.IntField;
+import org.apache.lucene.index.IndexableField;
+import org.elasticsearch.index.fielddata.FieldDataType;
+import org.elasticsearch.index.fielddata.IndexFieldData;
+import org.elasticsearch.index.fielddata.fieldcomparator.ByteValuesComparatorSource;
+import org.elasticsearch.index.fielddata.fieldcomparator.SortMode;
+import org.elasticsearch.index.fielddata.plain.ByteArrayIndexFieldData;
+
+/**
+ */
+public class ByteNestedSortingTests extends AbstractNumberNestedSortingTests {
+
+    @Override
+    protected FieldDataType getFieldDataType() {
+        return new FieldDataType("byte");
+    }
+
+    @Override
+    protected IndexFieldData.XFieldComparatorSource createInnerFieldComparator(String fieldName, SortMode sortMode) {
+        ByteArrayIndexFieldData fieldData = getForField(fieldName);
+        return new ByteValuesComparatorSource(fieldData, null, sortMode);
+    }
+
+    @Override
+    protected IndexableField createField(String name, int value, Field.Store store) {
+        return new IntField(name, value, store);
+    }
+}

src/test/java/org/elasticsearch/test/unit/index/search/nested/DoubleNestedSortingTests.java

@@ -0,0 +1,83 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon 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.test.unit.index.search.nested;
+
+import org.apache.lucene.document.DoubleField;
+import org.apache.lucene.document.Field;
+import org.apache.lucene.index.IndexableField;
+import org.apache.lucene.search.*;
+import org.apache.lucene.search.join.ScoreMode;
+import org.apache.lucene.search.join.ToParentBlockJoinQuery;
+import org.elasticsearch.common.lucene.search.NotFilter;
+import org.elasticsearch.common.lucene.search.XFilteredQuery;
+import org.elasticsearch.index.fielddata.FieldDataType;
+import org.elasticsearch.index.fielddata.IndexFieldData;
+import org.elasticsearch.index.fielddata.fieldcomparator.DoubleValuesComparatorSource;
+import org.elasticsearch.index.fielddata.fieldcomparator.SortMode;
+import org.elasticsearch.index.fielddata.plain.DoubleArrayIndexFieldData;
+import org.elasticsearch.index.search.nested.NestedFieldComparatorSource;
+
+import java.io.IOException;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+
+/**
+ */
+public class DoubleNestedSortingTests extends AbstractNumberNestedSortingTests {
+
+    @Override
+    protected FieldDataType getFieldDataType() {
+        return new FieldDataType("double");
+    }
+
+    @Override
+    protected IndexFieldData.XFieldComparatorSource createInnerFieldComparator(String fieldName, SortMode sortMode) {
+        DoubleArrayIndexFieldData fieldData = getForField(fieldName);
+        return new DoubleValuesComparatorSource(fieldData, null, sortMode);
+    }
+
+    @Override
+    protected IndexableField createField(String name, int value, Field.Store store) {
+        return new DoubleField(name, value, store);
+    }
+
+    protected void assertAvgScoreMode(Filter parentFilter, IndexSearcher searcher, IndexFieldData.XFieldComparatorSource innerFieldComparator) throws IOException {
+        SortMode sortMode = SortMode.AVG;
+        Filter childFilter = new NotFilter(parentFilter);
+        NestedFieldComparatorSource nestedComparatorSource = new NestedFieldComparatorSource(sortMode, innerFieldComparator, parentFilter, childFilter);
+        Query query = new ToParentBlockJoinQuery(new XFilteredQuery(new MatchAllDocsQuery(), childFilter), new CachingWrapperFilter(parentFilter), ScoreMode.None);
+        Sort sort = new Sort(new SortField("field2", nestedComparatorSource));
+        TopDocs topDocs = searcher.search(query, 5, sort);
+        assertThat(topDocs.totalHits, equalTo(7));
+        assertThat(topDocs.scoreDocs.length, equalTo(5));
+        assertThat(topDocs.scoreDocs[0].doc, equalTo(11));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[0]).fields[0]).intValue(), equalTo(2));
+        assertThat(topDocs.scoreDocs[1].doc, equalTo(7));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[1]).fields[0]).intValue(), equalTo(2));
+        assertThat(topDocs.scoreDocs[2].doc, equalTo(3));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[2]).fields[0]).intValue(), equalTo(3));
+        assertThat(topDocs.scoreDocs[3].doc, equalTo(15));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[3]).fields[0]).intValue(), equalTo(3));
+        assertThat(topDocs.scoreDocs[4].doc, equalTo(19));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[4]).fields[0]).intValue(), equalTo(3));
+    }
+
+}

src/test/java/org/elasticsearch/test/unit/index/search/nested/FloatNestedSortingTests.java

@@ -0,0 +1,83 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon 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.test.unit.index.search.nested;
+
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.FloatField;
+import org.apache.lucene.index.IndexableField;
+import org.apache.lucene.search.*;
+import org.apache.lucene.search.join.ScoreMode;
+import org.apache.lucene.search.join.ToParentBlockJoinQuery;
+import org.elasticsearch.common.lucene.search.NotFilter;
+import org.elasticsearch.common.lucene.search.XFilteredQuery;
+import org.elasticsearch.index.fielddata.FieldDataType;
+import org.elasticsearch.index.fielddata.IndexFieldData;
+import org.elasticsearch.index.fielddata.fieldcomparator.FloatValuesComparatorSource;
+import org.elasticsearch.index.fielddata.fieldcomparator.SortMode;
+import org.elasticsearch.index.fielddata.plain.FloatArrayIndexFieldData;
+import org.elasticsearch.index.search.nested.NestedFieldComparatorSource;
+
+import java.io.IOException;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+
+/**
+ */
+public class FloatNestedSortingTests extends AbstractNumberNestedSortingTests {
+
+    @Override
+    protected FieldDataType getFieldDataType() {
+        return new FieldDataType("float");
+    }
+
+    @Override
+    protected IndexFieldData.XFieldComparatorSource createInnerFieldComparator(String fieldName, SortMode sortMode) {
+        FloatArrayIndexFieldData fieldData = getForField(fieldName);
+        return new FloatValuesComparatorSource(fieldData, null, sortMode);
+    }
+
+    @Override
+    protected IndexableField createField(String name, int value, Field.Store store) {
+        return new FloatField(name, value, store);
+    }
+
+    protected void assertAvgScoreMode(Filter parentFilter, IndexSearcher searcher, IndexFieldData.XFieldComparatorSource innerFieldComparator) throws IOException {
+        SortMode sortMode = SortMode.AVG;
+        Filter childFilter = new NotFilter(parentFilter);
+        NestedFieldComparatorSource nestedComparatorSource = new NestedFieldComparatorSource(sortMode, innerFieldComparator, parentFilter, childFilter);
+        Query query = new ToParentBlockJoinQuery(new XFilteredQuery(new MatchAllDocsQuery(), childFilter), new CachingWrapperFilter(parentFilter), ScoreMode.None);
+        Sort sort = new Sort(new SortField("field2", nestedComparatorSource));
+        TopDocs topDocs = searcher.search(query, 5, sort);
+        assertThat(topDocs.totalHits, equalTo(7));
+        assertThat(topDocs.scoreDocs.length, equalTo(5));
+        assertThat(topDocs.scoreDocs[0].doc, equalTo(11));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[0]).fields[0]).intValue(), equalTo(2));
+        assertThat(topDocs.scoreDocs[1].doc, equalTo(7));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[1]).fields[0]).intValue(), equalTo(2));
+        assertThat(topDocs.scoreDocs[2].doc, equalTo(3));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[2]).fields[0]).intValue(), equalTo(3));
+        assertThat(topDocs.scoreDocs[3].doc, equalTo(15));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[3]).fields[0]).intValue(), equalTo(3));
+        assertThat(topDocs.scoreDocs[4].doc, equalTo(19));
+        assertThat(((Number) ((FieldDoc) topDocs.scoreDocs[4]).fields[0]).intValue(), equalTo(3));
+    }
+
+}

src/test/java/org/elasticsearch/test/unit/index/search/nested/IntegerNestedSortingTests.java

@@ -0,0 +1,50 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon 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.test.unit.index.search.nested;
+
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.IntField;
+import org.apache.lucene.index.IndexableField;
+import org.elasticsearch.index.fielddata.FieldDataType;
+import org.elasticsearch.index.fielddata.IndexFieldData;
+import org.elasticsearch.index.fielddata.fieldcomparator.IntValuesComparatorSource;
+import org.elasticsearch.index.fielddata.fieldcomparator.SortMode;
+import org.elasticsearch.index.fielddata.plain.IntArrayIndexFieldData;
+
+/**
+ */
+public class IntegerNestedSortingTests extends AbstractNumberNestedSortingTests {
+
+    @Override
+    protected FieldDataType getFieldDataType() {
+        return new FieldDataType("int");
+    }
+
+    @Override
+    protected IndexFieldData.XFieldComparatorSource createInnerFieldComparator(String fieldName, SortMode sortMode) {
+        IntArrayIndexFieldData fieldData = getForField(fieldName);
+        return new IntValuesComparatorSource(fieldData, null, sortMode);
+    }
+
+    @Override
+    protected IndexableField createField(String name, int value, Field.Store store) {
+        return new IntField(name, value, store);
+    }
+}

src/test/java/org/elasticsearch/test/unit/index/search/nested/LongNestedSortingTests.java

@@ -0,0 +1,51 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon 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.test.unit.index.search.nested;
+
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.LongField;
+import org.apache.lucene.index.IndexableField;
+import org.elasticsearch.index.fielddata.FieldDataType;
+import org.elasticsearch.index.fielddata.IndexFieldData;
+import org.elasticsearch.index.fielddata.fieldcomparator.LongValuesComparatorSource;
+import org.elasticsearch.index.fielddata.fieldcomparator.SortMode;
+import org.elasticsearch.index.fielddata.plain.LongArrayIndexFieldData;
+
+/**
+ */
+public class LongNestedSortingTests extends AbstractNumberNestedSortingTests {
+
+    @Override
+    protected FieldDataType getFieldDataType() {
+        return new FieldDataType("long");
+    }
+
+    @Override
+    protected IndexFieldData.XFieldComparatorSource createInnerFieldComparator(String fieldName, SortMode sortMode) {
+        LongArrayIndexFieldData fieldData = getForField(fieldName);
+        return new LongValuesComparatorSource(fieldData, null, sortMode);
+    }
+
+    @Override
+    protected IndexableField createField(String name, int value, Field.Store store) {
+        return new LongField(name, value, store);
+    }
+
+}

src/test/java/org/elasticsearch/test/unit/index/search/nested/NestedSortingTests.java

@@ -1,3 +1,23 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon 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.test.unit.index.search.nested;
 
 import org.apache.lucene.document.Document;
@@ -31,7 +51,7 @@ import static org.hamcrest.Matchers.equalTo;
 
 /**
  */
-public class NestedFieldComparatorTests extends AbstractFieldDataTests {
+public class NestedSortingTests extends AbstractFieldDataTests {
 
     @Override
     protected FieldDataType getFieldDataType() {

src/test/java/org/elasticsearch/test/unit/index/search/nested/ShortNestedSortingTests.java

@@ -0,0 +1,50 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon 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.test.unit.index.search.nested;
+
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.IntField;
+import org.apache.lucene.index.IndexableField;
+import org.elasticsearch.index.fielddata.FieldDataType;
+import org.elasticsearch.index.fielddata.IndexFieldData;
+import org.elasticsearch.index.fielddata.fieldcomparator.ShortValuesComparatorSource;
+import org.elasticsearch.index.fielddata.fieldcomparator.SortMode;
+import org.elasticsearch.index.fielddata.plain.ShortArrayIndexFieldData;
+
+/**
+ */
+public class ShortNestedSortingTests extends AbstractNumberNestedSortingTests {
+
+    @Override
+    protected FieldDataType getFieldDataType() {
+        return new FieldDataType("short");
+    }
+
+    @Override
+    protected IndexFieldData.XFieldComparatorSource createInnerFieldComparator(String fieldName, SortMode sortMode) {
+        ShortArrayIndexFieldData fieldData = getForField(fieldName);
+        return new ShortValuesComparatorSource(fieldData, null, sortMode);
+    }
+
+    @Override
+    protected IndexableField createField(String name, int value, Field.Store store) {
+        return new IntField(name, value, store);
+    }
+}