[DATAES-433] Support join datatype.

Original PR: #485

src/main/java/org/springframework/data/elasticsearch/annotations/JoinTypeRelation.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2020 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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.springframework.data.elasticsearch.annotations;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Inherited;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * @author Subhobrata Dey
+ * @since 4.1
+ */
+@Documented
+@Retention(RetentionPolicy.RUNTIME)
+@Target(ElementType.ANNOTATION_TYPE)
+public @interface JoinTypeRelation {
+
+    String parent();
+
+    String[] children();
+}

src/main/java/org/springframework/data/elasticsearch/annotations/JoinTypeRelations.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2020 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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.springframework.data.elasticsearch.annotations;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Inherited;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * @author Subhobrata Dey
+ * @since 4.1
+ */
+@Documented
+@Retention(RetentionPolicy.RUNTIME)
+@Target(ElementType.FIELD)
+@Inherited
+public @interface JoinTypeRelations {
+
+    JoinTypeRelation[] relations();
+}

src/main/java/org/springframework/data/elasticsearch/core/AbstractElasticsearchTemplate.java

@@ -39,6 +39,8 @@ import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.data.convert.EntityReader;
 import org.springframework.data.elasticsearch.BulkFailureException;
+import org.springframework.data.elasticsearch.core.join.JoinField;
+import org.springframework.data.elasticsearch.annotations.JoinTypeRelations;
 import org.springframework.data.elasticsearch.core.convert.ElasticsearchConverter;
 import org.springframework.data.elasticsearch.core.convert.MappingElasticsearchConverter;
 import org.springframework.data.elasticsearch.core.document.Document;
@@ -532,6 +534,22 @@ public abstract class AbstractElasticsearchTemplate implements ElasticsearchOper
 		return null;
 	}
 
+	@Nullable
+	private String getEntityRouting(Object entity) {
+		ElasticsearchPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entity.getClass());
+		ElasticsearchPersistentProperty joinProperty = persistentEntity.getJoinFieldProperty();
+
+		if (joinProperty != null) {
+			Object joinField = persistentEntity.getPropertyAccessor(entity).getProperty(joinProperty);
+			if (joinField != null && JoinField.class.isAssignableFrom(joinField.getClass())
+				&& ((JoinField<?>) joinField).getParent() != null) {
+				return elasticsearchConverter.convertId(((JoinField<?>) joinField).getParent());
+			}
+		}
+
+		return null;
+	}
+
 	@Nullable
 	private Long getEntityVersion(Object entity) {
 		ElasticsearchPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entity.getClass());
@@ -581,6 +599,11 @@ public abstract class AbstractElasticsearchTemplate implements ElasticsearchOper
 			// version cannot be used together with seq_no and primary_term
 			builder.withVersion(getEntityVersion(entity));
 		}
+
+		String routing = getEntityRouting(entity);
+		if (routing != null) {
+			builder.withRouting(routing);
+		}
 		return builder.build();
 	}
 

src/main/java/org/springframework/data/elasticsearch/core/RequestFactory.java

@@ -743,6 +743,10 @@ class RequestFactory {
 			deleteByQueryRequest.setScroll(TimeValue.timeValueMillis(query.getScrollTime().toMillis()));
 		}
 
+		if (query.getRoute() != null) {
+			deleteByQueryRequest.setRouting(query.getRoute());
+		}
+
 		return deleteByQueryRequest;
 	}
 
@@ -798,6 +802,10 @@ class RequestFactory {
 			source.setScroll(TimeValue.timeValueMillis(query.getScrollTime().toMillis()));
 		}
 
+		if (query.getRoute() != null) {
+			source.setRouting(query.getRoute());
+		}
+
 		return requestBuilder;
 	}
 	// endregion
@@ -888,6 +896,10 @@ class RequestFactory {
 			indexRequest.setIfPrimaryTerm(query.getPrimaryTerm());
 		}
 
+		if (query.getRouting() != null) {
+			indexRequest.routing(query.getRouting());
+		}
+
 		return indexRequest;
 	}
 
@@ -926,6 +938,9 @@ class RequestFactory {
 		if (query.getPrimaryTerm() != null) {
 			indexRequestBuilder.setIfPrimaryTerm(query.getPrimaryTerm());
 		}
+		if (query.getRouting() != null) {
+			indexRequestBuilder.setRouting(query.getRouting());
+		}
 
 		return indexRequestBuilder;
 	}

src/main/java/org/springframework/data/elasticsearch/core/convert/MappingElasticsearchConverter.java

@@ -44,6 +44,7 @@ import org.springframework.data.elasticsearch.ElasticsearchException;
 import org.springframework.data.elasticsearch.annotations.ScriptedField;
 import org.springframework.data.elasticsearch.core.document.Document;
 import org.springframework.data.elasticsearch.core.document.SearchDocument;
+import org.springframework.data.elasticsearch.core.join.JoinField;
 import org.springframework.data.elasticsearch.core.mapping.ElasticsearchPersistentEntity;
 import org.springframework.data.elasticsearch.core.mapping.ElasticsearchPersistentProperty;
 import org.springframework.data.elasticsearch.core.mapping.ElasticsearchPersistentPropertyConverter;
@@ -710,6 +711,13 @@ public class MappingElasticsearchConverter
 			if (container.equals(type) && type.getType().equals(actualType)) {
 				return false;
 			}
+
+			if (container.getRawTypeInformation().equals(type)) {
+				Class<?> containerClass = container.getRawTypeInformation().getType();
+				if (containerClass.equals(JoinField.class) && type.getType().equals(actualType)) {
+					return false;
+				}
+			}
 		}
 
 		return !conversions.isSimpleType(type.getType()) && !type.isCollectionLike()

src/main/java/org/springframework/data/elasticsearch/core/index/MappingBuilder.java

@@ -40,6 +40,8 @@ import org.springframework.data.elasticsearch.annotations.Field;
 import org.springframework.data.elasticsearch.annotations.FieldType;
 import org.springframework.data.elasticsearch.annotations.GeoPointField;
 import org.springframework.data.elasticsearch.annotations.InnerField;
+import org.springframework.data.elasticsearch.annotations.JoinTypeRelation;
+import org.springframework.data.elasticsearch.annotations.JoinTypeRelations;
 import org.springframework.data.elasticsearch.annotations.Mapping;
 import org.springframework.data.elasticsearch.annotations.MultiField;
 import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
@@ -47,6 +49,7 @@ import org.springframework.data.elasticsearch.core.ResourceUtil;
 import org.springframework.data.elasticsearch.core.completion.Completion;
 import org.springframework.data.elasticsearch.core.convert.ElasticsearchConverter;
 import org.springframework.data.elasticsearch.core.geo.GeoPoint;
+import org.springframework.data.elasticsearch.core.join.JoinField;
 import org.springframework.data.elasticsearch.core.mapping.ElasticsearchPersistentEntity;
 import org.springframework.data.elasticsearch.core.mapping.ElasticsearchPersistentProperty;
 import org.springframework.data.mapping.MappingException;
@@ -73,6 +76,7 @@ import com.fasterxml.jackson.databind.ObjectMapper;
  * @author Petr Kukral
  * @author Peter-Josef Meisch
  * @author Xiao Yu
+ * @author Subhobrata Dey
  */
 public class MappingBuilder {
 
@@ -93,8 +97,11 @@ public class MappingBuilder {
 	private static final String TYPE_DYNAMIC = "dynamic";
 	private static final String TYPE_VALUE_KEYWORD = "keyword";
 	private static final String TYPE_VALUE_GEO_POINT = "geo_point";
+	private static final String TYPE_VALUE_JOIN = "join";
 	private static final String TYPE_VALUE_COMPLETION = "completion";
 
+	private static final String JOIN_TYPE_RELATIONS = "relations";
+
 	private static final Logger logger = LoggerFactory.getLogger(ElasticsearchRestTemplate.class);
 
 	private final ElasticsearchConverter elasticsearchConverter;
@@ -212,6 +219,10 @@ public class MappingBuilder {
 			return;
 		}
 
+		if (isJoinFieldProperty(property)) {
+			addJoinFieldMapping(builder, property);
+		}
+
 		Field fieldAnnotation = property.findAnnotation(Field.class);
 		boolean isCompletionProperty = isCompletionProperty(property);
 		boolean isNestedOrObjectProperty = isNestedOrObjectProperty(property);
@@ -336,6 +347,36 @@ public class MappingBuilder {
 		builder.endObject();
 	}
 
+	private void addJoinFieldMapping(XContentBuilder builder,
+									 ElasticsearchPersistentProperty property) throws IOException {
+		JoinTypeRelation[] joinTypeRelations = property.getRequiredAnnotation(JoinTypeRelations.class).relations();
+
+		if (joinTypeRelations.length == 0) {
+			logger.warn("Property {}s type is JoinField but its annotation JoinTypeRelation is " + //
+							"not properly maintained", //
+					property.getFieldName());
+			return;
+		}
+		builder.startObject(property.getFieldName());
+
+		builder.field(FIELD_PARAM_TYPE, TYPE_VALUE_JOIN);
+
+		builder.startObject(JOIN_TYPE_RELATIONS);
+
+		for (JoinTypeRelation joinTypeRelation: joinTypeRelations) {
+			String parent = joinTypeRelation.parent();
+			String[] children = joinTypeRelation.children();
+
+			if (children.length > 1) {
+				builder.array(parent, children);
+			} else if (children.length == 1) {
+				builder.field(parent, children[0]);
+			}
+		}
+		builder.endObject();
+		builder.endObject();
+	}
+
 	/**
 	 * Add mapping for @MultiField annotation
 	 *
@@ -423,6 +464,10 @@ public class MappingBuilder {
 		return property.getActualType() == GeoPoint.class || property.isAnnotationPresent(GeoPointField.class);
 	}
 
+	private boolean isJoinFieldProperty(ElasticsearchPersistentProperty property) {
+		return property.getActualType() == JoinField.class;
+	}
+
 	private boolean isCompletionProperty(ElasticsearchPersistentProperty property) {
 		return property.getActualType() == Completion.class;
 	}

src/main/java/org/springframework/data/elasticsearch/core/join/JoinField.java

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2020 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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.springframework.data.elasticsearch.core.join;
+
+import org.springframework.lang.Nullable;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * @author Subhobrata Dey
+ * @since 4.1
+ */
+public class JoinField<ID> {
+
+    private final String name;
+
+    @Nullable private ID parent;
+
+    public JoinField() {
+        this("default", null);
+    }
+
+    public JoinField(String name) {
+        this(name, null);
+    }
+
+    public JoinField(String name, @Nullable ID parent) {
+        this.name = name;
+        this.parent = parent;
+    }
+
+    public void setParent(@Nullable ID parent) {
+        this.parent = parent;
+    }
+
+    @Nullable
+    public ID getParent() {
+        return parent;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public Map<String, Object> getAsMap() {
+        Map<String, Object> joinMap = new HashMap<>();
+        joinMap.put("name", getName());
+        joinMap.put("parent", getParent());
+
+        return Collections.unmodifiableMap(joinMap);
+    }
+}

src/main/java/org/springframework/data/elasticsearch/core/mapping/ElasticsearchPersistentEntity.java

@@ -18,6 +18,7 @@ package org.springframework.data.elasticsearch.core.mapping;
 import org.elasticsearch.index.VersionType;
 import org.springframework.data.elasticsearch.annotations.Field;
 import org.springframework.data.elasticsearch.core.document.Document;
+import org.springframework.data.elasticsearch.core.join.JoinField;
 import org.springframework.data.elasticsearch.core.query.SeqNoPrimaryTerm;
 import org.springframework.data.mapping.PersistentEntity;
 import org.springframework.lang.Nullable;
@@ -33,6 +34,7 @@ import org.springframework.lang.Nullable;
  * @author Ivan Greene
  * @author Peter-Josef Meisch
  * @author Roman Puchkovskiy
+ * @author Subhobrata Dey
  */
 public interface ElasticsearchPersistentEntity<T> extends PersistentEntity<T, ElasticsearchPersistentProperty> {
 
@@ -119,6 +121,15 @@ public interface ElasticsearchPersistentEntity<T> extends PersistentEntity<T, El
 	 */
 	boolean hasSeqNoPrimaryTermProperty();
 
+	/**
+	 * Returns whether the {@link ElasticsearchPersistentEntity} has a {@link JoinField} property. If this call
+	 * returns {@literal true}, {@link #getJoinFieldProperty()} will return a non-{@literal null} value.
+	 *
+	 * @return false when {@link ElasticsearchPersistentEntity} does not define a JoinField property.
+	 * @since 4.1
+	 */
+	boolean hasJoinFieldProperty();
+
 	/**
 	 * Returns the {@link SeqNoPrimaryTerm} property of the {@link ElasticsearchPersistentEntity}. Can be {@literal null}
 	 * in case no such property is available on the entity.
@@ -130,6 +141,17 @@ public interface ElasticsearchPersistentEntity<T> extends PersistentEntity<T, El
 	@Nullable
 	ElasticsearchPersistentProperty getSeqNoPrimaryTermProperty();
 
+	/**
+	 * Returns the {@link JoinField} property of the {@link ElasticsearchPersistentEntity}. Can be {@literal null}
+	 * in case no such property is available on the entity.
+	 *
+	 * @return the {@link JoinField} {@link ElasticsearchPersistentProperty} of the {@link PersistentEntity} or
+	 *         {@literal null} if not defined.
+	 * @since 4.1
+	 */
+	@Nullable
+	ElasticsearchPersistentProperty getJoinFieldProperty();
+
 	/**
 	 * Returns the {@link SeqNoPrimaryTerm} property of the {@link ElasticsearchPersistentEntity} or throws an
 	 * IllegalStateException in case no such property is available on the entity.

src/main/java/org/springframework/data/elasticsearch/core/mapping/SimpleElasticsearchPersistentEntity.java

@@ -26,6 +26,7 @@ import org.slf4j.LoggerFactory;
 import org.springframework.data.elasticsearch.annotations.Parent;
 import org.springframework.data.elasticsearch.annotations.Setting;
 import org.springframework.data.elasticsearch.core.document.Document;
+import org.springframework.data.elasticsearch.core.join.JoinField;
 import org.springframework.data.mapping.MappingException;
 import org.springframework.data.mapping.PropertyHandler;
 import org.springframework.data.mapping.model.BasicPersistentEntity;
@@ -67,6 +68,7 @@ public class SimpleElasticsearchPersistentEntity<T> extends BasicPersistentEntit
 	@Deprecated private @Nullable ElasticsearchPersistentProperty parentIdProperty;
 	private @Nullable ElasticsearchPersistentProperty scoreProperty;
 	private @Nullable ElasticsearchPersistentProperty seqNoPrimaryTermProperty;
+	private @Nullable ElasticsearchPersistentProperty joinFieldProperty;
 	private @Nullable String settingPath;
 	private @Nullable VersionType versionType;
 	private boolean createIndexAndMapping;
@@ -230,6 +232,19 @@ public class SimpleElasticsearchPersistentEntity<T> extends BasicPersistentEntit
 				warnAboutBothSeqNoPrimaryTermAndVersionProperties();
 			}
 		}
+
+		if (property.getActualType() == JoinField.class) {
+			ElasticsearchPersistentProperty joinProperty = this.joinFieldProperty;
+
+			if (joinProperty != null) {
+				throw new MappingException(String.format(
+						"Attempt to add Join property %s but already have property %s registered "
+								+ "as Join property. Check your entity configuration!",
+						property.getField(), joinProperty.getField()));
+			}
+
+			this.joinFieldProperty = property;
+		}
 	}
 
 	private void warnAboutBothSeqNoPrimaryTermAndVersionProperties() {
@@ -273,12 +288,22 @@ public class SimpleElasticsearchPersistentEntity<T> extends BasicPersistentEntit
 		return seqNoPrimaryTermProperty != null;
 	}
 
+	@Override
+	public boolean hasJoinFieldProperty() {
+		return joinFieldProperty != null;
+	}
+
 	@Override
 	@Nullable
 	public ElasticsearchPersistentProperty getSeqNoPrimaryTermProperty() {
 		return seqNoPrimaryTermProperty;
 	}
 
+	@Override
+	public ElasticsearchPersistentProperty getJoinFieldProperty() {
+		return joinFieldProperty;
+	}
+
 	// region SpEL handling
 	/**
 	 * resolves all the names in the IndexCoordinates object. If a name cannot be resolved, the original name is returned.

src/main/java/org/springframework/data/elasticsearch/core/query/IndexQuery.java

@@ -34,6 +34,7 @@ public class IndexQuery {
 	@Deprecated @Nullable private String parentId;
 	@Nullable private Long seqNo;
 	@Nullable private Long primaryTerm;
+	@Nullable private String routing;
 
 	@Nullable
 	public String getId() {
@@ -107,4 +108,13 @@ public class IndexQuery {
 	public void setPrimaryTerm(Long primaryTerm) {
 		this.primaryTerm = primaryTerm;
 	}
+
+	@Nullable
+	public String getRouting() {
+		return routing;
+	}
+
+	public void setRouting(@Nullable String routing) {
+		this.routing = routing;
+	}
 }

src/main/java/org/springframework/data/elasticsearch/core/query/IndexQueryBuilder.java

@@ -34,6 +34,7 @@ public class IndexQueryBuilder {
 	@Deprecated @Nullable private String parentId;
 	@Nullable private Long seqNo;
 	@Nullable private Long primaryTerm;
+	@Nullable private String routing;
 
 	public IndexQueryBuilder withId(String id) {
 		this.id = id;
@@ -67,6 +68,11 @@ public class IndexQueryBuilder {
 		return this;
 	}
 
+	public IndexQueryBuilder withRouting(@Nullable String routing) {
+		this.routing = routing;
+		return this;
+	}
+
 	public IndexQuery build() {
 		IndexQuery indexQuery = new IndexQuery();
 		indexQuery.setId(id);
@@ -76,6 +82,7 @@ public class IndexQueryBuilder {
 		indexQuery.setVersion(version);
 		indexQuery.setSeqNo(seqNo);
 		indexQuery.setPrimaryTerm(primaryTerm);
+		indexQuery.setRouting(routing);
 		return indexQuery;
 	}
 }

src/test/java/org/springframework/data/elasticsearch/Utils.java

@@ -15,11 +15,13 @@
  */
 package org.springframework.data.elasticsearch;
 
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.UUID;
 
 import org.elasticsearch.client.Client;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.join.ParentJoinPlugin;
 import org.elasticsearch.node.Node;
 import org.elasticsearch.node.NodeValidationException;
 import org.elasticsearch.transport.Netty4Plugin;
@@ -53,7 +55,7 @@ public class Utils {
 						.put("cluster.routing.allocation.disk.watermark.high", "1gb")//
 						.put("cluster.routing.allocation.disk.watermark.flood_stage", "1gb")//
 						.build(), //
-				Collections.singletonList(Netty4Plugin.class));
+				Arrays.asList(Netty4Plugin.class, ParentJoinPlugin.class));
 	}
 
 	public static Client getNodeClient() throws NodeValidationException {

src/test/java/org/springframework/data/elasticsearch/core/ElasticsearchRestTemplateTests.java

@@ -19,25 +19,28 @@ import static org.assertj.core.api.Assertions.*;
 import static org.springframework.data.elasticsearch.annotations.FieldType.*;
 import static org.springframework.data.elasticsearch.utils.IdGenerator.*;
 
-import lombok.Builder;
+import lombok.*;
-import lombok.Data;
-import lombok.val;
 
 import java.lang.Object;
-import java.util.Collections;
+import java.util.*;
-import java.util.HashMap;
+import java.util.function.Function;
-import java.util.Map;
+import java.util.stream.Collectors;
 
 import org.elasticsearch.action.support.ActiveShardCount;
 import org.elasticsearch.action.support.WriteRequest;
 import org.elasticsearch.action.update.UpdateRequest;
 import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.index.query.SimpleQueryStringBuilder;
 import org.junit.jupiter.api.Test;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.elasticsearch.UncategorizedElasticsearchException;
 import org.springframework.data.elasticsearch.annotations.Document;
 import org.springframework.data.elasticsearch.annotations.Field;
+import org.springframework.data.elasticsearch.annotations.JoinTypeRelation;
+import org.springframework.data.elasticsearch.annotations.JoinTypeRelations;
+import org.springframework.data.elasticsearch.core.join.JoinField;
 import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
+import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
 import org.springframework.data.elasticsearch.core.query.UpdateQuery;
 import org.springframework.data.elasticsearch.junit.jupiter.ElasticsearchRestTemplateConfiguration;
 import org.springframework.data.elasticsearch.junit.jupiter.SpringIntegrationTest;
@@ -118,5 +121,4 @@ public class ElasticsearchRestTemplateTests extends ElasticsearchTemplateTests {
 		assertThat(fetchSourceContext.includes()).containsExactlyInAnyOrder("incl");
 		assertThat(fetchSourceContext.excludes()).containsExactlyInAnyOrder("excl");
 	}
-
 }

src/test/java/org/springframework/data/elasticsearch/core/ElasticsearchTemplateTests.java

@@ -41,6 +41,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 
 import org.assertj.core.util.Lists;
@@ -49,6 +50,8 @@ import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.action.update.UpdateRequest;
 import org.elasticsearch.cluster.metadata.AliasMetadata;
 import org.elasticsearch.index.VersionType;
+import org.elasticsearch.index.query.SimpleQueryStringBuilder;
+import org.elasticsearch.join.query.ParentIdQueryBuilder;
 import org.elasticsearch.script.Script;
 import org.elasticsearch.script.ScriptType;
 import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
@@ -68,20 +71,17 @@ import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.domain.Sort.Order;
 import org.springframework.data.elasticsearch.ElasticsearchException;
-import org.springframework.data.elasticsearch.annotations.Document;
+import org.springframework.data.elasticsearch.annotations.*;
 import org.springframework.data.elasticsearch.annotations.Field;
-import org.springframework.data.elasticsearch.annotations.FieldType;
-import org.springframework.data.elasticsearch.annotations.InnerField;
-import org.springframework.data.elasticsearch.annotations.MultiField;
-import org.springframework.data.elasticsearch.annotations.Score;
-import org.springframework.data.elasticsearch.annotations.ScriptedField;
 import org.springframework.data.elasticsearch.core.geo.GeoPoint;
 import org.springframework.data.elasticsearch.core.index.AliasAction;
 import org.springframework.data.elasticsearch.core.index.AliasActionParameters;
 import org.springframework.data.elasticsearch.core.index.AliasActions;
 import org.springframework.data.elasticsearch.core.index.AliasData;
+import org.springframework.data.elasticsearch.core.join.JoinField;
 import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
 import org.springframework.data.elasticsearch.core.query.*;
+import org.springframework.data.elasticsearch.core.query.Query;
 import org.springframework.data.util.StreamUtils;
 import org.springframework.lang.Nullable;
 
@@ -117,9 +117,10 @@ public abstract class ElasticsearchTemplateTests {
 	private static final String INDEX_3_NAME = "test-index-3";
 
 	protected final IndexCoordinates index = IndexCoordinates.of(INDEX_NAME_SAMPLE_ENTITY);
+	protected static final String INDEX_NAME_JOIN_SAMPLE_ENTITY = "test-index-sample-join-template";
 
 	@Autowired protected ElasticsearchOperations operations;
-	private IndexOperations indexOperations;
+	protected IndexOperations indexOperations;
 
 	@BeforeEach
 	public void before() {
@@ -136,6 +137,10 @@ public abstract class ElasticsearchTemplateTests {
 		IndexOperations indexOpsSearchHitsEntity = operations.indexOps(SearchHitsEntity.class);
 		indexOpsSearchHitsEntity.create();
 		indexOpsSearchHitsEntity.putMapping(SearchHitsEntity.class);
+
+		IndexOperations indexOpsJoinEntity = operations.indexOps(ElasticsearchRestTemplateTests.SampleJoinEntity.class);
+		indexOpsJoinEntity.create();
+		indexOpsJoinEntity.putMapping(ElasticsearchRestTemplateTests.SampleJoinEntity.class);
 	}
 
 	@AfterEach
@@ -158,6 +163,7 @@ public abstract class ElasticsearchTemplateTests {
 		operations.indexOps(HighlightEntity.class).delete();
 		operations.indexOps(OptimisticEntity.class).delete();
 		operations.indexOps(OptimisticAndVersionedEntity.class).delete();
+		operations.indexOps(IndexCoordinates.of(INDEX_NAME_JOIN_SAMPLE_ENTITY)).delete();
 	}
 
 	@Test // DATAES-106
@@ -171,7 +177,6 @@ public abstract class ElasticsearchTemplateTests {
 		IndexQuery indexQuery = getIndexQuery(sampleEntity);
 		operations.index(indexQuery, index);
 		indexOperations.refresh();
-		;
 		CriteriaQuery criteriaQuery = new CriteriaQuery(new Criteria());
 
 		// when
@@ -3302,6 +3307,135 @@ public abstract class ElasticsearchTemplateTests {
 		operations.save(forEdit);
 	}
 
+	@Test
+	void shouldSupportCRUDOpsForEntityWithJoinFields() throws Exception {
+		String qId1 = java.util.UUID.randomUUID().toString();
+		String qId2 = java.util.UUID.randomUUID().toString();
+		String aId1 = java.util.UUID.randomUUID().toString();
+		String aId2 = java.util.UUID.randomUUID().toString();
+
+		shouldSaveEntityWithJoinFields(qId1, qId2, aId1, aId2);
+		shouldUpdateEntityWithJoinFields(qId1, qId2, aId1, aId2);
+		shouldDeleteEntityWithJoinFields(qId2, aId2);
+	}
+
+	void shouldSaveEntityWithJoinFields(String qId1, String qId2, String aId1, String aId2) throws Exception {
+		SampleJoinEntity sampleQuestionEntity1 = new SampleJoinEntity();
+		sampleQuestionEntity1.setUuid(qId1);
+		sampleQuestionEntity1.setText("This is a question");
+
+		JoinField<String> myQJoinField1 = new JoinField<>("question");
+		sampleQuestionEntity1.setMyJoinField(myQJoinField1);
+
+		SampleJoinEntity sampleQuestionEntity2 = new SampleJoinEntity();
+		sampleQuestionEntity2.setUuid(qId2);
+		sampleQuestionEntity2.setText("This is another question");
+
+		JoinField<String> myQJoinField2 = new JoinField<>("question");
+		sampleQuestionEntity2.setMyJoinField(myQJoinField2);
+
+		SampleJoinEntity sampleAnswerEntity1 = new SampleJoinEntity();
+		sampleAnswerEntity1.setUuid(aId1);
+		sampleAnswerEntity1.setText("This is an answer");
+
+		JoinField<String> myAJoinField1 = new JoinField<>("answer");
+		myAJoinField1.setParent(qId1);
+		sampleAnswerEntity1.setMyJoinField(myAJoinField1);
+
+		SampleJoinEntity sampleAnswerEntity2 = new SampleJoinEntity();
+		sampleAnswerEntity2.setUuid(aId2);
+		sampleAnswerEntity2.setText("This is another answer");
+
+		JoinField<String> myAJoinField2 = new JoinField<>("answer");
+		myAJoinField2.setParent(qId1);
+		sampleAnswerEntity2.setMyJoinField(myAJoinField2);
+
+		operations.save(Arrays.asList(sampleQuestionEntity1, sampleQuestionEntity2,
+				sampleAnswerEntity1, sampleAnswerEntity2), IndexCoordinates.of(INDEX_NAME_JOIN_SAMPLE_ENTITY));
+		indexOperations.refresh();
+		Thread.sleep(5000);
+
+		SearchHits<SampleJoinEntity> hits = operations.search(new NativeSearchQueryBuilder().withQuery(
+				new ParentIdQueryBuilder("answer", qId1))
+				.build(), SampleJoinEntity.class);
+
+		List<String> hitIds = hits.getSearchHits().stream().map(new Function<SearchHit<SampleJoinEntity>, String>() {
+			@Override
+			public String apply(SearchHit<SampleJoinEntity> sampleJoinEntitySearchHit) {
+				return sampleJoinEntitySearchHit.getId();
+			}
+		}).collect(Collectors.toList());
+
+		assertThat(hitIds.size()).isEqualTo(2);
+		assertThat(hitIds.containsAll(Arrays.asList(aId1, aId2))).isTrue();
+	}
+
+	void shouldUpdateEntityWithJoinFields(String qId1, String qId2, String aId1, String aId2) throws Exception {
+		org.springframework.data.elasticsearch.core.document.Document document = org.springframework.data.elasticsearch.core.document.Document
+				.create();
+		document.put("myJoinField", new JoinField<>("answer", qId2).getAsMap());
+		UpdateQuery updateQuery = UpdateQuery.builder(aId2) //
+				.withDocument(document) //
+				.withRouting(qId2)
+				.build();
+
+		List<UpdateQuery> queries = new ArrayList<>();
+		queries.add(updateQuery);
+
+		// when
+		operations.bulkUpdate(queries, IndexCoordinates.of(INDEX_NAME_JOIN_SAMPLE_ENTITY));
+		indexOperations.refresh();
+		Thread.sleep(5000);
+
+		SearchHits<SampleJoinEntity> updatedHits = operations.search(new NativeSearchQueryBuilder().withQuery(
+				new ParentIdQueryBuilder("answer", qId2))
+				.build(), SampleJoinEntity.class);
+
+		List<String> hitIds = updatedHits.getSearchHits().stream().map(new Function<SearchHit<SampleJoinEntity>, String>() {
+			@Override
+			public String apply(SearchHit<SampleJoinEntity> sampleJoinEntitySearchHit) {
+				return sampleJoinEntitySearchHit.getId();
+			}
+		}).collect(Collectors.toList());
+		assertThat(hitIds.size()).isEqualTo(1);
+		assertThat(hitIds.get(0)).isEqualTo(aId2);
+
+		updatedHits = operations.search(new NativeSearchQueryBuilder().withQuery(
+				new ParentIdQueryBuilder("answer", qId1))
+				.build(), SampleJoinEntity.class);
+
+		hitIds = updatedHits.getSearchHits().stream().map(new Function<SearchHit<SampleJoinEntity>, String>() {
+			@Override
+			public String apply(SearchHit<SampleJoinEntity> sampleJoinEntitySearchHit) {
+				return sampleJoinEntitySearchHit.getId();
+			}
+		}).collect(Collectors.toList());
+		assertThat(hitIds.size()).isEqualTo(1);
+		assertThat(hitIds.get(0)).isEqualTo(aId1);
+	}
+
+	void shouldDeleteEntityWithJoinFields(String qId2, String aId2) throws Exception {
+		Query query = new NativeSearchQueryBuilder()
+				.withQuery(new ParentIdQueryBuilder("answer", qId2))
+				.withRoute(qId2)
+				.build();
+		operations.delete(query, SampleJoinEntity.class, IndexCoordinates.of(INDEX_NAME_JOIN_SAMPLE_ENTITY));
+		indexOperations.refresh();
+		Thread.sleep(5000);
+
+		SearchHits<SampleJoinEntity> deletedHits = operations.search(new NativeSearchQueryBuilder().withQuery(
+				new ParentIdQueryBuilder("answer", qId2))
+				.build(), SampleJoinEntity.class);
+
+		List<String> hitIds = deletedHits.getSearchHits().stream().map(new Function<SearchHit<SampleJoinEntity>, String>() {
+			@Override
+			public String apply(SearchHit<SampleJoinEntity> sampleJoinEntitySearchHit) {
+				return sampleJoinEntitySearchHit.getId();
+			}
+		}).collect(Collectors.toList());
+		assertThat(hitIds.size()).isEqualTo(0);
+	}
+
 	protected RequestFactory getRequestFactory() {
 		return ((AbstractElasticsearchTemplate) operations).getRequestFactory();
 	}
@@ -3482,4 +3616,18 @@ public abstract class ElasticsearchTemplateTests {
 		private SeqNoPrimaryTerm seqNoPrimaryTerm;
 		@Version private Long version;
 	}
+
+	@Data
+	@NoArgsConstructor
+	@AllArgsConstructor
+	@Builder
+	@Document(indexName = INDEX_NAME_JOIN_SAMPLE_ENTITY)
+	static class SampleJoinEntity {
+		@Id @Field(type = Keyword) private String uuid;
+		@JoinTypeRelations(relations = {
+				@JoinTypeRelation(parent = "question", children = {"answer"})
+		})
+		private JoinField<String> myJoinField;
+		@Field(type = Text) private String text;
+	}
 }