Allow field types to optimize phrase prefix queries

This change adds a way to customize how phrase prefix queries should be created
on field types. The match phrase prefix query is exposed in field types in order
to allow optimizations based on the options set on the field.
For instance the text field uses the configured prefix field (if available) to
build a span near that mixes the original field and the prefix field on the last
position.
This change also contains a small refactoring of the match/multi_match query that
simplifies the interactions between the builders.

Closes elastic#31921

server/src/main/java/org/elasticsearch/search/suggest/completion/CompletionSuggestion.java

@@ -20,6 +20,7 @@
 
 import org.apache.lucene.analysis.CharArraySet;
 import org.apache.lucene.search.ScoreDoc;
+import org.apache.lucene.util.PriorityQueue;
 import org.elasticsearch.Version;
 import org.elasticsearch.common.ParseField;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -42,7 +43,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
-import java.util.PriorityQueue;
 import java.util.Set;
 
 import static org.elasticsearch.common.xcontent.XContentParserUtils.ensureExpectedToken;
@@ -138,29 +138,34 @@ public static CompletionSuggestion fromXContent(XContentParser parser, String na
         return suggestion;
     }
 
-    private static class SuggestionRef implements Iterator<Entry.Option>, Comparable<SuggestionRef> {
-        final Iterator<Entry.Option> it;
-        Entry.Option current;
-
-        private SuggestionRef(Iterator<Entry.Option> it) {
-            assert it.hasNext();
-            this.it = it;
-            this.current = it.next();
+    private static final class OptionPriorityQueue extends PriorityQueue<ShardOptions> {
+        OptionPriorityQueue(int maxSize) {
+            super(maxSize);
         }
 
         @Override
-        public int compareTo(SuggestionRef o) {
-            return COMPARATOR.compare(current, o.current);
+        protected boolean lessThan(ShardOptions a, ShardOptions b) {
+            return COMPARATOR.compare(a.current, b.current) < 0;
         }
+    }
 
-        @Override
-        public boolean hasNext() {
-            return it.hasNext();
+    private static class ShardOptions {
+        final Iterator<Entry.Option> optionsIterator;
+        Entry.Option current;
+
+        private ShardOptions(Iterator<Entry.Option> optionsIterator) {
+            assert optionsIterator.hasNext();
+            this.optionsIterator = optionsIterator;
+            this.current = optionsIterator.next();
         }
 
-        @Override
-        public Entry.Option next() {
-            return current = it.next();
+        boolean advanceToNextOption() {
+            if (optionsIterator.hasNext()) {
+                current = optionsIterator.next();
+                return true;
+            } else {
+                return false;
+            }
         }
     }
 
@@ -182,33 +187,33 @@ public static CompletionSuggestion reduceTo(List<Suggest.Suggestion<Entry>> toRe
                 // combine suggestion entries from participating shards on the coordinating node
                 // the global top <code>size</code> entries are collected from the shard results
                 // using a priority queue
-                PriorityQueue<SuggestionRef> pq = new PriorityQueue(leader.getSize());
+                OptionPriorityQueue pq = new OptionPriorityQueue(toReduce.size());
                 for (Suggest.Suggestion<Entry> suggestion : toReduce) {
                     assert suggestion.getName().equals(name) : "name should be identical across all suggestions";
                     Iterator<Entry.Option> it = ((CompletionSuggestion) suggestion).getOptions().iterator();
                     if (it.hasNext()) {
-                        pq.add(new SuggestionRef(it));
+                        pq.add(new ShardOptions(it));
                     }
-
                 }
                 // Dedup duplicate suggestions (based on the surface form) if skip duplicates is activated
                 final CharArraySet seenSurfaceForms = leader.skipDuplicates ? new CharArraySet(leader.getSize(), false) : null;
                 final Entry entry = new Entry(leaderEntry.getText(), leaderEntry.getOffset(), leaderEntry.getLength());
                 final List<Entry.Option> options = entry.getOptions();
-                while (pq.isEmpty() == false) {
-                    final SuggestionRef ref = pq.poll();
-                    final Entry.Option current = ref.current;
-                    if (ref.hasNext()) {
-                        ref.next();
-                        pq.add(ref);
-                    }
-                    if (leader.skipDuplicates &&
-                            seenSurfaceForms.add(current.getText().toString()) == false) {
-                        continue;
+                while (pq.size() > 0) {
+                    ShardOptions top = pq.top();
+                    Entry.Option current = top.current;
+                    if (top.advanceToNextOption()) {
+                        pq.updateTop();
+                    } else {
+                        //options exhausted for this shard
+                        pq.pop();
                     }
-                    options.add(current);
-                    if (options.size() >= size) {
-                        break;
+                    if (leader.skipDuplicates == false ||
+                            seenSurfaceForms.add(current.getText().toString())) {
+                        options.add(current);
+                        if (options.size() >= size) {
+                            break;
+                        }
                     }
                 }
                 final CompletionSuggestion suggestion = new CompletionSuggestion(leader.getName(), leader.getSize(), leader.skipDuplicates);

server/src/test/java/org/elasticsearch/search/suggest/completion/CompletionSuggestionTests.java

@@ -91,7 +91,7 @@ public void testToReduceWithDuplicates() {
             shardSuggestions.add(suggestion);
         }
         List<CompletionSuggestion.Entry.Option> expected = options.stream()
-            .sorted((o1, o2) -> COMPARATOR.compare(o1, o2))
+            .sorted(COMPARATOR)
             .distinct()
             .limit(size)
             .collect(Collectors.toList());