Search: Fix paging on strings sorted in ascending order.

For the comparator to work correctly, we need to give it the same value in
`setTopValue` as the value that it gave back in `value`.

Close #9136

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

@@ -91,17 +91,32 @@ public class BytesRefFieldComparatorSource extends IndexFieldData.XFieldComparat
                     }
                 }
                 
+                @Override
+                public void setScorer(Scorer scorer) {
+                    BytesRefFieldComparatorSource.this.setScorer(scorer);
+                }
+
                 public BytesRef value(int slot) {
                     // TODO: When serializing the response to the coordinating node, we lose the information about
                     // whether the comparator sorts missing docs first or last. We should fix it and let
                     // TopDocs.merge deal with it (it knows how to)
                     BytesRef value = super.value(slot);
                     if (value == null) {
+                        assert sortMissingFirst(missingValue) || sortMissingLast(missingValue);
                         value = missingBytes;
                     }
                     return value;
                 }
                 
+                public void setTopValue(BytesRef topValue) {
+                    // symetric of value(int): if we need to feed the comparator with <tt>null</tt>
+                    // if we overrode the value with MAX_TERM in value(int)
+                    if (topValue == missingBytes && (sortMissingFirst(missingValue) || sortMissingLast(missingValue))) {
+                        topValue = null;
+                    }
+                    super.setTopValue(topValue);
+                }
+
             };
         }
 

src/test/java/org/elasticsearch/search/scroll/SearchScrollTests.java

@@ -24,6 +24,7 @@ import org.elasticsearch.action.search.ClearScrollResponse;
 import org.elasticsearch.action.search.SearchRequestBuilder;
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.action.search.SearchType;
+import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.settings.ImmutableSettings;
 import org.elasticsearch.common.unit.TimeValue;
@@ -31,6 +32,7 @@ import org.elasticsearch.common.util.concurrent.UncategorizedExecutionException;
 import org.elasticsearch.index.query.QueryBuilders;
 import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.search.SearchHit;
+import org.elasticsearch.search.sort.FieldSortBuilder;
 import org.elasticsearch.search.sort.SortOrder;
 import org.elasticsearch.test.ElasticsearchIntegrationTest;
 import org.elasticsearch.test.hamcrest.ElasticsearchAssertions;
@@ -39,9 +41,15 @@ import org.junit.Test;
 import java.util.Map;
 
 import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
-import static org.elasticsearch.index.query.QueryBuilders.*;
-import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertThrows;
-import static org.hamcrest.Matchers.*;
+import static org.elasticsearch.index.query.QueryBuilders.matchAllQuery;
+import static org.elasticsearch.index.query.QueryBuilders.queryStringQuery;
+import static org.elasticsearch.index.query.QueryBuilders.termQuery;
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.*;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThan;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.notNullValue;
 
 /**
  *
@@ -448,4 +456,38 @@ public class SearchScrollTests extends ElasticsearchIntegrationTest {
 
         assertThrows(internalCluster().transportClient().prepareSearchScroll(searchResponse.getScrollId()), RestStatus.NOT_FOUND);
     }
+
+    @Test
+    public void testStringSortMissingAscTerminates() throws Exception {
+        assertAcked(prepareCreate("test")
+                .setSettings(ImmutableSettings.settingsBuilder().put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, 1).put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, 0))
+                .addMapping("test", "no_field", "type=string", "some_field", "type=string"));
+        client().prepareIndex("test", "test", "1").setSource("some_field", "test").get();
+        refresh();
+
+        SearchResponse response = client().prepareSearch("test")
+                .setTypes("test")
+                .addSort(new FieldSortBuilder("no_field").order(SortOrder.ASC).missing("_last"))
+                .setScroll("1m")
+                .get();
+        assertHitCount(response, 1);
+        assertSearchHits(response, "1");
+
+        response = client().prepareSearchScroll(response.getScrollId()).get();
+        assertSearchResponse(response);
+        assertHitCount(response, 1);
+        assertNoSearchHits(response);
+
+        response = client().prepareSearch("test")
+                .setTypes("test")
+                .addSort(new FieldSortBuilder("no_field").order(SortOrder.ASC).missing("_first"))
+                .setScroll("1m")
+                .get();
+        assertHitCount(response, 1);
+        assertSearchHits(response, "1");
+
+        response = client().prepareSearchScroll(response.getScrollId()).get();
+        assertHitCount(response, 1);
+        assertThat(response.getHits().getHits().length, equalTo(0));
+    }
 }

src/test/java/org/elasticsearch/test/hamcrest/ElasticsearchAssertions.java

@@ -145,6 +145,10 @@ public class ElasticsearchAssertions {
         assertVersionSerializable(searchResponse);
     }
 
+    public static void assertNoSearchHits(SearchResponse searchResponse) {
+        assertEquals(0, searchResponse.getHits().getHits().length);
+    }
+
     public static void assertSearchHits(SearchResponse searchResponse, String... ids) {
         String shardStatus = formatShardStatus(searchResponse);