Introduce ParentJoinFieldMapper, a field mapper that creates parent/c…

…hild relation within documents of the same index (#24978)

* Introduce ParentJoinFieldMapper, a field mapper that creates parent/child relation within documents of the same index

This change adds a new field mapper named ParentJoinFieldMapper. This mapper is a replacement for the ParentFieldMapper but instead of using the types in the mapping
it uses an internal field to materialize parent/child relation within a single index.
This change also adds a fetch sub phase that automatically retrieves the join name (parent or child name) and the parent id for child documents in the response hit fields.
The compatibility with `has_parent`, `has_child` queries and `children` agg will be added in a follow up.

Relates #20257

modules/parent-join/src/main/java/org/elasticsearch/join/ParentJoinPlugin.java

@@ -20,18 +20,24 @@
 package org.elasticsearch.join;
 
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.index.mapper.Mapper;
 import org.elasticsearch.join.aggregations.ChildrenAggregationBuilder;
 import org.elasticsearch.join.aggregations.InternalChildren;
+import org.elasticsearch.join.fetch.ParentJoinFieldSubFetchPhase;
+import org.elasticsearch.join.mapper.ParentJoinFieldMapper;
 import org.elasticsearch.join.query.HasChildQueryBuilder;
 import org.elasticsearch.join.query.HasParentQueryBuilder;
+import org.elasticsearch.plugins.MapperPlugin;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.plugins.SearchPlugin;
+import org.elasticsearch.search.fetch.FetchSubPhase;
 
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 
-public class ParentJoinPlugin extends Plugin implements SearchPlugin {
+public class ParentJoinPlugin extends Plugin implements SearchPlugin, MapperPlugin {
     public ParentJoinPlugin(Settings settings) {}
 
     @Override
@@ -50,5 +56,13 @@ public List<AggregationSpec> getAggregations() {
         );
     }
 
+    @Override
+    public Map<String, Mapper.TypeParser> getMappers() {
+        return Collections.singletonMap(ParentJoinFieldMapper.CONTENT_TYPE, new ParentJoinFieldMapper.TypeParser());
+    }
 
+    @Override
+    public List<FetchSubPhase> getFetchSubPhases(FetchPhaseConstructionContext context) {
+        return Collections.singletonList(new ParentJoinFieldSubFetchPhase());
+    }
 }

modules/parent-join/src/main/java/org/elasticsearch/join/fetch/ParentJoinFieldSubFetchPhase.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.join.fetch;
+
+import org.apache.lucene.index.LeafReader;
+import org.apache.lucene.index.SortedDocValues;
+import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.ExceptionsHelper;
+import org.elasticsearch.Version;
+import org.elasticsearch.common.collect.Tuple;
+import org.elasticsearch.index.mapper.DocumentMapper;
+import org.elasticsearch.index.mapper.FieldMapper;
+import org.elasticsearch.join.mapper.ParentJoinFieldMapper;
+import org.elasticsearch.search.SearchHitField;
+import org.elasticsearch.search.fetch.FetchSubPhase;
+import org.elasticsearch.search.internal.SearchContext;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * A sub fetch phase that retrieves the join name and the parent id for each document containing
+ * a {@link ParentJoinFieldMapper} field.
+ */
+public final class ParentJoinFieldSubFetchPhase implements FetchSubPhase {
+    @Override
+    public void hitExecute(SearchContext context, HitContext hitContext) {
+        if (context.storedFieldsContext() != null && context.storedFieldsContext().fetchFields() == false) {
+            return;
+        }
+        if (context.mapperService().getIndexSettings().getIndexVersionCreated().before(Version.V_6_0_0_alpha2)) {
+            return;
+        }
+        DocumentMapper docMapper = context.mapperService().documentMapper(hitContext.hit().getType());
+        Tuple<String, String> joinField = null;
+        Tuple<String, String> parentField = null;
+        for (FieldMapper fieldMapper : docMapper.mappers()) {
+            if (fieldMapper instanceof ParentJoinFieldMapper) {
+                String joinName = getSortedDocValue(fieldMapper.name(), hitContext.reader(), hitContext.docId());
+                if (joinName != null) {
+                    ParentJoinFieldMapper joinFieldMapper = (ParentJoinFieldMapper) fieldMapper;
+                    joinField = new Tuple<>(fieldMapper.name(), joinName);
+                    // we retrieve the parent id only for children.
+                    FieldMapper parentMapper = joinFieldMapper.getParentIdFieldMapper(joinName, false);
+                    if (parentMapper != null) {
+                        String parent = getSortedDocValue(parentMapper.name(), hitContext.reader(), hitContext.docId());
+                        parentField = new Tuple<>(parentMapper.name(), parent);
+                    }
+                    break;
+                }
+            }
+        }
+
+        if (joinField == null) {
+            // hit has no join field.
+            return;
+        }
+
+        Map<String, SearchHitField> fields = hitContext.hit().fieldsOrNull();
+        if (fields == null) {
+            fields = new HashMap<>();
+            hitContext.hit().fields(fields);
+        }
+        fields.put(joinField.v1(), new SearchHitField(joinField.v1(), Collections.singletonList(joinField.v2())));
+        if (parentField != null) {
+            fields.put(parentField.v1(), new SearchHitField(parentField.v1(), Collections.singletonList(parentField.v2())));
+        }
+    }
+
+    private String getSortedDocValue(String field, LeafReader reader, int docId) {
+        try {
+            SortedDocValues docValues = reader.getSortedDocValues(field);
+            if (docValues == null || docValues.advanceExact(docId) == false) {
+                return null;
+            }
+            int ord = docValues.ordValue();
+            BytesRef joinName = docValues.lookupOrd(ord);
+            return joinName.utf8ToString();
+        } catch (IOException e) {
+            throw ExceptionsHelper.convertToElastic(e);
+        }
+    }
+}

modules/parent-join/src/main/java/org/elasticsearch/join/mapper/ParentIdFieldMapper.java

@@ -0,0 +1,174 @@
+/*
+ * 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.join.mapper;
+
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.SortedDocValuesField;
+import org.apache.lucene.index.IndexOptions;
+import org.apache.lucene.index.IndexableField;
+import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.common.lucene.Lucene;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.index.fielddata.IndexFieldData;
+import org.elasticsearch.index.fielddata.plain.DocValuesIndexFieldData;
+import org.elasticsearch.index.mapper.FieldMapper;
+import org.elasticsearch.index.mapper.MappedFieldType;
+import org.elasticsearch.index.mapper.Mapper;
+import org.elasticsearch.index.mapper.ParseContext;
+import org.elasticsearch.index.mapper.StringFieldType;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * A field mapper used internally by the {@link ParentJoinFieldMapper} to index
+ * the value that link documents in the index (parent _id or _id if the document is a parent).
+ */
+public final class ParentIdFieldMapper extends FieldMapper {
+    static final String CONTENT_TYPE = "parent";
+
+    static class Defaults {
+        public static final MappedFieldType FIELD_TYPE = new ParentIdFieldType();
+
+        static {
+            FIELD_TYPE.setTokenized(false);
+            FIELD_TYPE.setOmitNorms(true);
+            FIELD_TYPE.setHasDocValues(true);
+            FIELD_TYPE.setIndexOptions(IndexOptions.DOCS);
+            FIELD_TYPE.freeze();
+        }
+    }
+
+    static class Builder extends FieldMapper.Builder<Builder, ParentIdFieldMapper> {
+        private final String parent;
+        private final Set<String> children;
+
+        Builder(String name, String parent, Set<String> children) {
+            super(name, Defaults.FIELD_TYPE, Defaults.FIELD_TYPE);
+            builder = this;
+            this.parent = parent;
+            this.children = children;
+        }
+
+        public Set<String> getChildren() {
+            return children;
+        }
+
+        @Override
+        public ParentIdFieldMapper build(BuilderContext context) {
+            fieldType.setName(name);
+            return new ParentIdFieldMapper(name, parent, children, fieldType, context.indexSettings());
+        }
+    }
+
+    public static final class ParentIdFieldType extends StringFieldType {
+        public ParentIdFieldType() {
+            setIndexAnalyzer(Lucene.KEYWORD_ANALYZER);
+            setSearchAnalyzer(Lucene.KEYWORD_ANALYZER);
+        }
+
+        protected ParentIdFieldType(ParentIdFieldType ref) {
+            super(ref);
+        }
+
+        public ParentIdFieldType clone() {
+            return new ParentIdFieldType(this);
+        }
+
+        @Override
+        public String typeName() {
+            return CONTENT_TYPE;
+        }
+
+        @Override
+        public IndexFieldData.Builder fielddataBuilder() {
+            failIfNoDocValues();
+            return new DocValuesIndexFieldData.Builder();
+        }
+
+        @Override
+        public Object valueForDisplay(Object value) {
+            if (value == null) {
+                return null;
+            }
+            BytesRef binaryValue = (BytesRef) value;
+            return binaryValue.utf8ToString();
+        }
+    }
+
+    private final String parentName;
+    private Set<String> children;
+
+    protected ParentIdFieldMapper(String simpleName,
+                                  String parentName,
+                                  Set<String> children,
+                                  MappedFieldType fieldType,
+                                  Settings indexSettings) {
+        super(simpleName, fieldType, Defaults.FIELD_TYPE, indexSettings, MultiFields.empty(), null);
+        this.parentName = parentName;
+        this.children = children;
+    }
+
+    @Override
+    protected ParentIdFieldMapper clone() {
+        return (ParentIdFieldMapper) super.clone();
+    }
+
+    /**
+     * Returns the parent name associated with this mapper.
+     */
+    public String getParentName() {
+        return parentName;
+    }
+
+    /**
+     * Returns the children names associated with this mapper.
+     */
+    public Collection<String> getChildren() {
+        return children;
+    }
+
+    @Override
+    protected void parseCreateField(ParseContext context, List<IndexableField> fields) throws IOException {
+        if (context.externalValueSet() == false) {
+            throw new IllegalStateException("external value not set");
+        }
+        String refId = (String) context.externalValue();
+        BytesRef binaryValue = new BytesRef(refId);
+        Field field = new Field(fieldType().name(), binaryValue, fieldType());
+        fields.add(field);
+        fields.add(new SortedDocValuesField(fieldType().name(), binaryValue));
+    }
+
+
+    @Override
+    protected void doMerge(Mapper mergeWith, boolean updateAllTypes) {
+        super.doMerge(mergeWith, updateAllTypes);
+        ParentIdFieldMapper parentMergeWith = (ParentIdFieldMapper) mergeWith;
+        this.children = parentMergeWith.children;
+    }
+
+    @Override
+    protected String contentType() {
+        return CONTENT_TYPE;
+    }
+}