From aade2f6d6341dc67f2df7d5c5e5f32836d549069 Mon Sep 17 00:00:00 2001
From: Jim Ferenczi <jim.ferenczi@elastic.co>
Date: Thu, 28 Sep 2017 16:24:05 +0200
Subject: [PATCH] Change ParentFieldSubFetchPhase to create doc values iterator
 once per segment (#26815)

---
 .../subphase/ParentFieldSubFetchPhase.java    | 69 ++++++++++++++-----
 1 file changed, 52 insertions(+), 17 deletions(-)

diff --git a/core/src/main/java/org/elasticsearch/search/fetch/subphase/ParentFieldSubFetchPhase.java b/core/src/main/java/org/elasticsearch/search/fetch/subphase/ParentFieldSubFetchPhase.java
index 93fc9ccb1f6..6015e3c9021 100644
--- a/core/src/main/java/org/elasticsearch/search/fetch/subphase/ParentFieldSubFetchPhase.java
+++ b/core/src/main/java/org/elasticsearch/search/fetch/subphase/ParentFieldSubFetchPhase.java
@@ -20,43 +20,79 @@
 package org.elasticsearch.search.fetch.subphase;
 
 import org.apache.lucene.index.LeafReader;
+import org.apache.lucene.index.LeafReaderContext;
+import org.apache.lucene.index.ReaderUtil;
 import org.apache.lucene.index.SortedDocValues;
 import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.common.document.DocumentField;
+import org.elasticsearch.index.mapper.MapperService;
 import org.elasticsearch.index.mapper.ParentFieldMapper;
+import org.elasticsearch.search.SearchHit;
 import org.elasticsearch.search.fetch.FetchSubPhase;
 import org.elasticsearch.search.internal.SearchContext;
 
 import java.io.IOException;
+import java.util.Arrays;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Map;
+import java.util.Set;
 
 public final class ParentFieldSubFetchPhase implements FetchSubPhase {
-
     @Override
-    public void hitExecute(SearchContext context, HitContext hitContext) {
+    public void hitsExecute(SearchContext context, SearchHit[] hits) throws IOException {
         if (context.storedFieldsContext() != null && context.storedFieldsContext().fetchFields() == false) {
             return ;
         }
-        ParentFieldMapper parentFieldMapper = context.mapperService().documentMapper(hitContext.hit().getType()).parentFieldMapper();
-        if (parentFieldMapper.active() == false) {
-            return;
+
+        hits = hits.clone(); // don't modify the incoming hits
+        Arrays.sort(hits, Comparator.comparingInt(SearchHit::docId));
+
+        MapperService mapperService = context.mapperService();
+        Set<String> parentFields = new HashSet<>();
+        for (SearchHit hit : hits) {
+            ParentFieldMapper parentFieldMapper = mapperService.documentMapper(hit.getType()).parentFieldMapper();
+            if (parentFieldMapper.active()) {
+                parentFields.add(parentFieldMapper.name());
+            }
         }
 
-        String parentId = getParentId(parentFieldMapper, hitContext.reader(), hitContext.docId());
-        if (parentId == null) {
-            // hit has no _parent field. Can happen for nested inner hits if parent hit is a p/c document.
-            return;
+        int lastReaderId = -1;
+        Map<String, SortedDocValues> docValuesMap = new HashMap<>();
+        for (SearchHit hit : hits) {
+            ParentFieldMapper parentFieldMapper = mapperService.documentMapper(hit.getType()).parentFieldMapper();
+            if (parentFieldMapper.active() == false) {
+                continue;
+            }
+            int readerId = ReaderUtil.subIndex(hit.docId(), context.searcher().getIndexReader().leaves());
+            LeafReaderContext subReaderContext = context.searcher().getIndexReader().leaves().get(readerId);
+            if (lastReaderId != readerId) {
+                docValuesMap.clear();
+                for (String field : parentFields) {
+                    docValuesMap.put(field, subReaderContext.reader().getSortedDocValues(field));
+                }
+                lastReaderId = readerId;
+            }
+            int docId = hit.docId() - subReaderContext.docBase;
+            SortedDocValues values = docValuesMap.get(parentFieldMapper.name());
+            if (values != null && values.advanceExact(docId)) {
+                BytesRef binaryValue = values.binaryValue();
+                String value = binaryValue.length > 0 ? binaryValue.utf8ToString() : null;
+                if (value == null) {
+                    // hit has no _parent field. Can happen for nested inner hits if parent hit is a p/c document.
+                    continue;
+                }
+                Map<String, DocumentField> fields = hit.fieldsOrNull();
+                if (fields == null) {
+                    fields = new HashMap<>();
+                    hit.fields(fields);
+                }
+                fields.put(ParentFieldMapper.NAME, new DocumentField(ParentFieldMapper.NAME, Collections.singletonList(value)));
+            }
         }
-
-        Map<String, DocumentField> fields = hitContext.hit().fieldsOrNull();
-        if (fields == null) {
-            fields = new HashMap<>();
-            hitContext.hit().fields(fields);
-        }
-        fields.put(ParentFieldMapper.NAME, new DocumentField(ParentFieldMapper.NAME, Collections.singletonList(parentId)));
     }
 
     public static String getParentId(ParentFieldMapper fieldMapper, LeafReader reader, int docId) {
@@ -72,5 +108,4 @@ public final class ParentFieldSubFetchPhase implements FetchSubPhase {
             throw ExceptionsHelper.convertToElastic(e);
         }
     }
-
 }