Resolve all persistent entities for a persistent property.

Original pull request: #2394.
Closes #2390.

src/main/java/org/springframework/data/mapping/context/AbstractMappingContext.java

@@ -34,7 +34,6 @@ import java.util.stream.StreamSupport;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-
 import org.springframework.beans.BeanUtils;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.InitializingBean;
@@ -116,7 +115,8 @@ public abstract class AbstractMappingContext<E extends MutablePersistentEntity<?
 		this.persistentPropertyPathFactory = new PersistentPropertyPathFactory<>(this);
 
 		EntityInstantiators instantiators = new EntityInstantiators();
-		PersistentPropertyAccessorFactory accessorFactory = NativeDetector.inNativeImage() ? BeanWrapperPropertyAccessorFactory.INSTANCE
+		PersistentPropertyAccessorFactory accessorFactory = NativeDetector.inNativeImage()
+				? BeanWrapperPropertyAccessorFactory.INSTANCE
 				: new ClassGeneratingPropertyAccessorFactory();
 
 		this.persistentPropertyAccessorFactory = new InstantiationAwarePropertyAccessorFactory(accessorFactory,
@@ -485,7 +485,8 @@ public abstract class AbstractMappingContext<E extends MutablePersistentEntity<?
 		if (simpleTypeHolder.isSimpleType(type.getType())) {
 			return false;
 		}
-		if(NullableWrapperConverters.supports(type.getType())) {
+
+		if (NullableWrapperConverters.supports(type.getType())) {
 			return false;
 		}
 
@@ -564,6 +565,7 @@ public abstract class AbstractMappingContext<E extends MutablePersistentEntity<?
 			if (shouldSkipOverrideProperty(property)) {
 				return;
 			}
+
 			entity.addPersistentProperty(property);
 
 			if (property.isAssociation()) {
@@ -574,18 +576,8 @@ public abstract class AbstractMappingContext<E extends MutablePersistentEntity<?
 				return;
 			}
 
-			StreamSupport.stream(property.getPersistentEntityTypes().spliterator(), false)
-					.map(it -> {
-						if(it.isNullableWrapper()) {
-							return it.getActualType();
-						}
-						return it;
-					})
-					.filter(it -> {
-
-						boolean shouldCreate = AbstractMappingContext.this.shouldCreatePersistentEntityFor(it);
-						return shouldCreate;
-					})
+			StreamSupport.stream(property.getPersistentEntityTypes().spliterator(), false) //
+					.filter(AbstractMappingContext.this::shouldCreatePersistentEntityFor) //
 					.forEach(AbstractMappingContext.this::addPersistentEntity);
 		}
 
@@ -668,7 +660,6 @@ public abstract class AbstractMappingContext<E extends MutablePersistentEntity<?
 		}
 	}
 
-
 	/**
 	 * Filter rejecting static fields as well as artificially introduced ones. See
 	 * {@link PersistentPropertyFilter#UNMAPPED_PROPERTIES} for details.

src/main/java/org/springframework/data/mapping/model/AbstractPersistentProperty.java

@@ -18,11 +18,16 @@ package org.springframework.data.mapping.model;
 import java.lang.reflect.Field;
 import java.lang.reflect.Method;
 import java.lang.reflect.Modifier;
+import java.util.Collection;
 import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.function.Supplier;
 
+import org.springframework.core.GenericTypeResolver;
 import org.springframework.data.mapping.Association;
 import org.springframework.data.mapping.PersistentEntity;
 import org.springframework.data.mapping.PersistentProperty;
@@ -64,6 +69,7 @@ public abstract class AbstractPersistentProperty<P extends PersistentProperty<P>
 
 	private final Lazy<Boolean> isAssociation;
 	private final Lazy<TypeInformation<?>> associationTargetType;
+	private final Lazy<Collection<TypeInformation<?>>> entityTypes;
 
 	private final Method getter;
 	private final Method setter;
@@ -111,6 +117,43 @@ public abstract class AbstractPersistentProperty<P extends PersistentProperty<P>
 		} else {
 			this.immutable = false;
 		}
+
+		this.entityTypes = Lazy.of(() -> collectEntityTypes(simpleTypeHolder, information, new LinkedHashSet<>()));
+	}
+
+	protected Set<TypeInformation<?>> collectEntityTypes(SimpleTypeHolder simpleTypeHolder, @Nullable TypeInformation<?> typeInformation, Set<TypeInformation<?>> entityTypes) {
+
+		if(typeInformation == null || entityTypes.contains(typeInformation) || simpleTypeHolder.isSimpleType(typeInformation.getType())) {
+			return entityTypes;
+		}
+
+		if(typeInformation.isMap()) {
+
+			collectEntityTypes(simpleTypeHolder, typeInformation.getComponentType(), entityTypes);
+			collectEntityTypes(simpleTypeHolder, typeInformation.getMapValueType(), entityTypes);
+			return entityTypes;
+		}
+
+		if(typeInformation.isCollectionLike()) {
+
+			collectEntityTypes(simpleTypeHolder, typeInformation.getComponentType(), entityTypes);
+			return entityTypes;
+		}
+
+		if(typeInformation.isNullableWrapper()) {
+
+			collectEntityTypes(simpleTypeHolder, typeInformation.getActualType(), entityTypes);
+			return entityTypes;
+		}
+
+		if(ASSOCIATION_TYPE != null && ASSOCIATION_TYPE.isAssignableFrom(typeInformation.getType())) {
+
+			entityTypes.add(getAssociationOrActualType());
+			return entityTypes;
+		}
+
+		entityTypes.add(typeInformation);
+		return entityTypes;
 	}
 
 	protected abstract Association<P> createAssociation();
@@ -167,13 +210,15 @@ public abstract class AbstractPersistentProperty<P extends PersistentProperty<P>
 	@Override
 	public Iterable<? extends TypeInformation<?>> getPersistentEntityTypes() {
 
+		if(isMap() || isCollectionLike()) {
+			return entityTypes.get();
+		}
+
 		if (!isEntity()) {
 			return Collections.emptySet();
 		}
 
-		TypeInformation<?> result = getAssociationTypeOr(() -> entityTypeInformation.getNullable());
-
-		return result != null ? Collections.singleton(result) : Collections.emptySet();
+		return entityTypes.get();
 	}
 
 	/*