Introduce AggregatePath.

AggregatePath replaces PersistentPropertyPathExtension.
It gets created and cached by the RelationalMappingContext, which should be more efficient and certainly looks nicer.

Closes #1525
Original pull request #1486

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/JdbcAggregateChangeExecutionContext.java

@@ -15,16 +15,7 @@
  */
 package org.springframework.data.jdbc.core;
 
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Set;
+import java.util.*;
 import java.util.function.BiConsumer;
 import java.util.stream.Collectors;
 
@@ -38,11 +29,12 @@ import org.springframework.data.jdbc.core.convert.JdbcIdentifierBuilder;
 import org.springframework.data.mapping.PersistentProperty;
 import org.springframework.data.mapping.PersistentPropertyPath;
 import org.springframework.data.mapping.PersistentPropertyPathAccessor;
-import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.relational.core.conversion.DbAction;
 import org.springframework.data.relational.core.conversion.DbActionExecutionResult;
 import org.springframework.data.relational.core.conversion.IdValueSource;
+import org.springframework.data.relational.core.mapping.AggregatePath;
 import org.springframework.data.relational.core.mapping.PersistentPropertyPathExtension;
+import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
 import org.springframework.data.relational.core.sql.LockMode;
@@ -65,7 +57,7 @@ class JdbcAggregateChangeExecutionContext {
 	private static final String UPDATE_FAILED = "Failed to update entity [%s]; Id [%s] not found in database";
 	private static final String UPDATE_FAILED_OPTIMISTIC_LOCKING = "Failed to update entity [%s]; The entity was updated since it was rea or it isn't in the database at all";
 
-	private final MappingContext<? extends RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty> context;
+	private final RelationalMappingContext context;
 	private final JdbcConverter converter;
 	private final DataAccessStrategy accessStrategy;
 
@@ -184,12 +176,11 @@ class JdbcAggregateChangeExecutionContext {
 		Object id = getParentId(action);
 
 		JdbcIdentifierBuilder identifier = JdbcIdentifierBuilder //
-				.forBackReferences(converter, new PersistentPropertyPathExtension(context, action.getPropertyPath()), id);
+				.forBackReferences(converter, context.getAggregatePath(action.getPropertyPath()), id);
 
 		for (Map.Entry<PersistentPropertyPath<RelationalPersistentProperty>, Object> qualifier : action.getQualifiers()
 				.entrySet()) {
-			identifier = identifier.withQualifier(new PersistentPropertyPathExtension(context, qualifier.getKey()),
-					qualifier.getValue());
+			identifier = identifier.withQualifier(context.getAggregatePath(qualifier.getKey()), qualifier.getValue());
 		}
 
 		return identifier.build();
@@ -197,26 +188,22 @@ class JdbcAggregateChangeExecutionContext {
 
 	private Object getParentId(DbAction.WithDependingOn<?> action) {
 
-		PersistentPropertyPathExtension path = new PersistentPropertyPathExtension(context, action.getPropertyPath());
-		PersistentPropertyPathExtension idPath = path.getIdDefiningParentPath();
-
-		DbAction.WithEntity<?> idOwningAction = getIdOwningAction(action, idPath);
+		DbAction.WithEntity<?> idOwningAction = getIdOwningAction(action, context.getAggregatePath(action.getPropertyPath()).getIdDefiningParentPath());
 
 		return getPotentialGeneratedIdFrom(idOwningAction);
 	}
 
-	private DbAction.WithEntity<?> getIdOwningAction(DbAction.WithEntity<?> action,
-			PersistentPropertyPathExtension idPath) {
+	private DbAction.WithEntity<?> getIdOwningAction(DbAction.WithEntity<?> action, AggregatePath idPath) {
 
 		if (!(action instanceof DbAction.WithDependingOn<?> withDependingOn)) {
 
-			Assert.state(idPath.getLength() == 0,
+			Assert.state(idPath.isRoot(),
 					"When the id path is not empty the id providing action should be of type WithDependingOn");
 
 			return action;
 		}
 
-		if (idPath.matches(withDependingOn.getPropertyPath())) {
+		if (idPath.equals(context.getAggregatePath(withDependingOn.getPropertyPath()))) {
 			return action;
 		}
 

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/BasicJdbcConverter.java

@@ -47,7 +47,8 @@ import org.springframework.data.mapping.model.SpELExpressionEvaluator;
 import org.springframework.data.mapping.model.SpELExpressionParameterValueProvider;
 import org.springframework.data.relational.core.conversion.BasicRelationalConverter;
 import org.springframework.data.relational.core.conversion.RelationalConverter;
-import org.springframework.data.relational.core.mapping.PersistentPropertyPathExtension;
+import org.springframework.data.relational.core.mapping.AggregatePath;
+import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
 import org.springframework.data.relational.core.sql.IdentifierProcessing;
@@ -86,14 +87,14 @@ public class BasicJdbcConverter extends BasicRelationalConverter implements Jdbc
 	 * Creates a new {@link BasicRelationalConverter} given {@link MappingContext} and a
 	 * {@link JdbcTypeFactory#unsupported() no-op type factory} throwing {@link UnsupportedOperationException} on type
 	 * creation. Use
-	 * {@link #BasicJdbcConverter(MappingContext, RelationResolver, CustomConversions, JdbcTypeFactory, IdentifierProcessing)}
+	 * {@link #BasicJdbcConverter(RelationalMappingContext, RelationResolver, CustomConversions, JdbcTypeFactory, IdentifierProcessing)}
 	 * (MappingContext, RelationResolver, JdbcTypeFactory)} to convert arrays and large objects into JDBC-specific types.
 	 *
 	 * @param context must not be {@literal null}.
 	 * @param relationResolver used to fetch additional relations from the database. Must not be {@literal null}.
 	 */
 	public BasicJdbcConverter(
-			MappingContext<? extends RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty> context,
+			RelationalMappingContext context,
 			RelationResolver relationResolver) {
 
 		super(context, new JdbcCustomConversions());
@@ -116,7 +117,7 @@ public class BasicJdbcConverter extends BasicRelationalConverter implements Jdbc
 	 * @since 2.0
 	 */
 	public BasicJdbcConverter(
-			MappingContext<? extends RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty> context,
+			RelationalMappingContext context,
 			RelationResolver relationResolver, CustomConversions conversions, JdbcTypeFactory typeFactory,
 			IdentifierProcessing identifierProcessing) {
 
@@ -300,12 +301,13 @@ public class BasicJdbcConverter extends BasicRelationalConverter implements Jdbc
 
 	@Override
 	public <T> T mapRow(RelationalPersistentEntity<T> entity, ResultSet resultSet, Object key) {
-		return new ReadingContext<T>(new PersistentPropertyPathExtension(getMappingContext(), entity),
+		return new ReadingContext<T>(getMappingContext().getAggregatePath( entity),
 				new ResultSetAccessor(resultSet), Identifier.empty(), key).mapRow();
 	}
 
+
 	@Override
-	public <T> T mapRow(PersistentPropertyPathExtension path, ResultSet resultSet, Identifier identifier, Object key) {
+	public <T> T mapRow(AggregatePath path, ResultSet resultSet, Identifier identifier, Object key) {
 		return new ReadingContext<T>(path, new ResultSetAccessor(resultSet), identifier, key).mapRow();
 	}
 
@@ -350,8 +352,8 @@ public class BasicJdbcConverter extends BasicRelationalConverter implements Jdbc
 
 		private final RelationalPersistentEntity<T> entity;
 
-		private final PersistentPropertyPathExtension rootPath;
-		private final PersistentPropertyPathExtension path;
+		private final AggregatePath rootPath;
+		private final AggregatePath path;
 		private final Identifier identifier;
 		private final Object key;
 
@@ -360,7 +362,7 @@ public class BasicJdbcConverter extends BasicRelationalConverter implements Jdbc
 		private final ResultSetAccessor accessor;
 
 		@SuppressWarnings("unchecked")
-		private ReadingContext(PersistentPropertyPathExtension rootPath, ResultSetAccessor accessor, Identifier identifier,
+		private ReadingContext(AggregatePath rootPath, ResultSetAccessor accessor, Identifier identifier,
 				Object key) {
 			RelationalPersistentEntity<T> entity = (RelationalPersistentEntity<T>) rootPath.getLeafEntity();
 
@@ -368,7 +370,7 @@ public class BasicJdbcConverter extends BasicRelationalConverter implements Jdbc
 
 			this.entity = entity;
 			this.rootPath = rootPath;
-			this.path = new PersistentPropertyPathExtension(getMappingContext(), this.entity);
+			this.path = getMappingContext().getAggregatePath( this.entity);
 			this.identifier = identifier;
 			this.key = key;
 			this.propertyValueProvider = new JdbcPropertyValueProvider(path, accessor);
@@ -376,10 +378,11 @@ public class BasicJdbcConverter extends BasicRelationalConverter implements Jdbc
 			this.accessor = accessor;
 		}
 
-		private ReadingContext(RelationalPersistentEntity<T> entity, PersistentPropertyPathExtension rootPath,
-				PersistentPropertyPathExtension path, Identifier identifier, Object key,
+		private ReadingContext(RelationalPersistentEntity<T> entity, AggregatePath rootPath,
+				AggregatePath path, Identifier identifier, Object key,
 				JdbcPropertyValueProvider propertyValueProvider,
 				JdbcBackReferencePropertyValueProvider backReferencePropertyValueProvider, ResultSetAccessor accessor) {
+
 			this.entity = entity;
 			this.rootPath = rootPath;
 			this.path = path;
@@ -393,7 +396,7 @@ public class BasicJdbcConverter extends BasicRelationalConverter implements Jdbc
 		private <S> ReadingContext<S> extendBy(RelationalPersistentProperty property) {
 			return new ReadingContext<>(
 					(RelationalPersistentEntity<S>) getMappingContext().getRequiredPersistentEntity(property.getActualType()),
-					rootPath.extendBy(property), path.extendBy(property), identifier, key,
+					rootPath.append(property), path.append(property), identifier, key,
 					propertyValueProvider.extendBy(property), backReferencePropertyValueProvider.extendBy(property), accessor);
 		}
 
@@ -453,10 +456,10 @@ public class BasicJdbcConverter extends BasicRelationalConverter implements Jdbc
 		private Iterable<Object> resolveRelation(@Nullable Object id, RelationalPersistentProperty property) {
 
 			Identifier identifier = id == null //
-					? this.identifier.withPart(rootPath.getQualifierColumn(), key, Object.class) //
-					: Identifier.of(rootPath.extendBy(property).getReverseColumnName(), id, Object.class);
+					? this.identifier.withPart(rootPath.getTableInfo().qualifierColumnInfo().name(), key, Object.class) //
+					: Identifier.of(rootPath.append(property).getTableInfo().reverseColumnInfo().name(), id, Object.class);
 
-			PersistentPropertyPath<? extends RelationalPersistentProperty> propertyPath = path.extendBy(property)
+			PersistentPropertyPath<? extends RelationalPersistentProperty> propertyPath = path.append(property)
 					.getRequiredPersistentPropertyPath();
 
 			return relationResolver.findAllByPath(identifier, propertyPath);

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/DefaultDataAccessStrategy.java

@@ -28,12 +28,12 @@ import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.mapping.PersistentPropertyPath;
 import org.springframework.data.relational.core.conversion.IdValueSource;
+import org.springframework.data.relational.core.mapping.AggregatePath;
 import org.springframework.data.relational.core.mapping.PersistentPropertyPathExtension;
 import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
 import org.springframework.data.relational.core.query.Query;
-import org.springframework.data.relational.core.sql.IdentifierProcessing;
 import org.springframework.data.relational.core.sql.LockMode;
 import org.springframework.data.relational.core.sql.SqlIdentifier;
 import org.springframework.jdbc.core.RowMapper;
@@ -297,8 +297,8 @@ public class DefaultDataAccessStrategy implements DataAccessStrategy {
 		Assert.notNull(identifier, "identifier must not be null");
 		Assert.notNull(propertyPath, "propertyPath must not be null");
 
-		PersistentPropertyPathExtension path = new PersistentPropertyPathExtension(context, propertyPath);
-		Class<?> actualType = path.getActualType();
+		AggregatePath path = context.getAggregatePath(propertyPath);
+		Class<?> actualType = path.getLeafEntity().getType();
 
 		String findAllByProperty = sql(actualType) //
 				.getFindAllByProperty(identifier, propertyPath);
@@ -339,8 +339,7 @@ public class DefaultDataAccessStrategy implements DataAccessStrategy {
 		String sqlQuery = sql(domainType).selectByQuery(query, parameterSource);
 
 		try {
-			return Optional.ofNullable(
-					operations.queryForObject(sqlQuery, parameterSource, getEntityRowMapper(domainType)));
+			return Optional.ofNullable(operations.queryForObject(sqlQuery, parameterSource, getEntityRowMapper(domainType)));
 		} catch (EmptyResultDataAccessException e) {
 			return Optional.empty();
 		}
@@ -394,14 +393,15 @@ public class DefaultDataAccessStrategy implements DataAccessStrategy {
 		return new EntityRowMapper<>(getRequiredPersistentEntity(domainType), converter);
 	}
 
-	private EntityRowMapper<?> getEntityRowMapper(PersistentPropertyPathExtension path, Identifier identifier) {
+	private EntityRowMapper<?> getEntityRowMapper(AggregatePath path, Identifier identifier) {
 		return new EntityRowMapper<>(path, converter, identifier);
 	}
 
-	private RowMapper<?> getMapEntityRowMapper(PersistentPropertyPathExtension path, Identifier identifier) {
+	private RowMapper<?> getMapEntityRowMapper(AggregatePath path, Identifier identifier) {
 
-		SqlIdentifier keyColumn = path.getQualifierColumn();
-		Assert.notNull(keyColumn, () -> "KeyColumn must not be null for " + path);
+		AggregatePath.ColumnInfo qualifierColumnInfo = path.getTableInfo().qualifierColumnInfo();
+		Assert.notNull(qualifierColumnInfo, () -> "Qualifier column must not be null for " + path);
+		SqlIdentifier keyColumn = qualifierColumnInfo.name();
 
 		return new MapEntityRowMapper<>(path, converter, identifier, keyColumn);
 	}

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/EntityRowMapper.java

@@ -17,6 +17,7 @@ package org.springframework.data.jdbc.core.convert;
 
 import java.sql.ResultSet;
 
+import org.springframework.data.relational.core.mapping.AggregatePath;
 import org.springframework.data.relational.core.mapping.PersistentPropertyPathExtension;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.jdbc.core.RowMapper;
@@ -35,13 +36,28 @@ import org.springframework.jdbc.core.RowMapper;
 public class EntityRowMapper<T> implements RowMapper<T> {
 
 	private final RelationalPersistentEntity<T> entity;
-	private final PersistentPropertyPathExtension path;
+	private final AggregatePath path;
 	private final JdbcConverter converter;
 	private final Identifier identifier;
 
+	/**
+	 *
+	 *
+	 * @deprecated use {@link EntityRowMapper#EntityRowMapper(AggregatePath, JdbcConverter, Identifier)} instead
+	 */
+	@Deprecated(since = "3.2", forRemoval = true)
 	@SuppressWarnings("unchecked")
 	public EntityRowMapper(PersistentPropertyPathExtension path, JdbcConverter converter, Identifier identifier) {
 
+		this.entity = (RelationalPersistentEntity<T>) path.getLeafEntity();
+		this.path = path.getAggregatePath();
+		this.converter = converter;
+		this.identifier = identifier;
+	}
+
+	@SuppressWarnings("unchecked")
+	public EntityRowMapper(AggregatePath path, JdbcConverter converter, Identifier identifier) {
+
 		this.entity = (RelationalPersistentEntity<T>) path.getLeafEntity();
 		this.path = path;
 		this.converter = converter;

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/JdbcBackReferencePropertyValueProvider.java

@@ -16,9 +16,8 @@
 package org.springframework.data.jdbc.core.convert;
 
 import org.springframework.data.mapping.model.PropertyValueProvider;
-import org.springframework.data.relational.core.mapping.PersistentPropertyPathExtension;
+import org.springframework.data.relational.core.mapping.AggregatePath;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
-import org.springframework.data.relational.core.sql.IdentifierProcessing;
 
 /**
  * {@link PropertyValueProvider} obtaining values from a {@link ResultSetAccessor}. For a given id property it provides
@@ -31,14 +30,14 @@ import org.springframework.data.relational.core.sql.IdentifierProcessing;
  */
 class JdbcBackReferencePropertyValueProvider implements PropertyValueProvider<RelationalPersistentProperty> {
 
-	private final PersistentPropertyPathExtension basePath;
+	private final AggregatePath basePath;
 	private final ResultSetAccessor resultSet;
 
 	/**
 	 * @param basePath path from the aggregate root relative to which all properties get resolved.
 	 * @param resultSet the {@link ResultSetAccessor} from which to obtain the actual values.
 	 */
-	JdbcBackReferencePropertyValueProvider(PersistentPropertyPathExtension basePath, ResultSetAccessor resultSet) {
+	JdbcBackReferencePropertyValueProvider(AggregatePath basePath, ResultSetAccessor resultSet) {
 
 		this.resultSet = resultSet;
 		this.basePath = basePath;
@@ -46,10 +45,10 @@ class JdbcBackReferencePropertyValueProvider implements PropertyValueProvider<Re
 
 	@Override
 	public <T> T getPropertyValue(RelationalPersistentProperty property) {
-		return (T) resultSet.getObject(basePath.extendBy(property).getReverseColumnNameAlias().getReference());
+		return (T) resultSet.getObject(basePath.append(property).getTableInfo().reverseColumnInfo().alias().getReference());
 	}
 
 	public JdbcBackReferencePropertyValueProvider extendBy(RelationalPersistentProperty property) {
-		return new JdbcBackReferencePropertyValueProvider(basePath.extendBy(property), resultSet);
+		return new JdbcBackReferencePropertyValueProvider(basePath.append(property), resultSet);
 	}
 }

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/JdbcConverter.java

@@ -19,8 +19,11 @@ import java.sql.ResultSet;
 import java.sql.SQLType;
 
 import org.springframework.data.jdbc.core.mapping.JdbcValue;
+import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.relational.core.conversion.RelationalConverter;
+import org.springframework.data.relational.core.mapping.AggregatePath;
 import org.springframework.data.relational.core.mapping.PersistentPropertyPathExtension;
+import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
 import org.springframework.data.util.TypeInformation;
@@ -67,8 +70,24 @@ public interface JdbcConverter extends RelationalConverter {
 	 * @param key primary key.
 	 * @param <T>
 	 * @return
+	 * @deprecated use {@link #mapRow(AggregatePath, ResultSet, Identifier, Object)} instead.
 	 */
-	<T> T mapRow(PersistentPropertyPathExtension path, ResultSet resultSet, Identifier identifier, Object key);
+	@Deprecated(since = "3.2", forRemoval = true)
+	default <T> T mapRow(PersistentPropertyPathExtension path, ResultSet resultSet, Identifier identifier, Object key){
+		return mapRow(path.getAggregatePath(), resultSet, identifier, key);
+	};
+
+	/**
+	 * Read the current row from {@link ResultSet} to an {@link AggregatePath#getLeafEntity()} entity}.
+	 *
+	 * @param path path to the owning property.
+	 * @param resultSet the {@link ResultSet} to read from.
+	 * @param identifier entity identifier.
+	 * @param key primary key.
+	 * @param <T>
+	 * @return
+	 */
+	<T> T mapRow(AggregatePath path, ResultSet resultSet, Identifier identifier, Object key);
 
 	/**
 	 * The type to be used to store this property in the database. Multidimensional arrays are unwrapped to reflect a
@@ -88,4 +107,7 @@ public interface JdbcConverter extends RelationalConverter {
 	 * @since 2.0
 	 */
 	SQLType getTargetSqlType(RelationalPersistentProperty property);
+
+	@Override
+	RelationalMappingContext getMappingContext();
 }

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/JdbcIdentifierBuilder.java

@@ -15,6 +15,7 @@
  */
 package org.springframework.data.jdbc.core.convert;
 
+import org.springframework.data.relational.core.mapping.AggregatePath;
 import org.springframework.data.relational.core.mapping.PersistentPropertyPathExtension;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -39,12 +40,23 @@ public class JdbcIdentifierBuilder {
 
 	/**
 	 * Creates ParentKeys with backreference for the given path and value of the parents id.
+	 *
+	 * @deprecated Use {@link #forBackReferences(JdbcConverter, AggregatePath, Object)} instead.
 	 */
+	@Deprecated(since = "3.2", forRemoval = true)
 	public static JdbcIdentifierBuilder forBackReferences(JdbcConverter converter, PersistentPropertyPathExtension path,
 			@Nullable Object value) {
+		return forBackReferences(converter, path.getAggregatePath(), value);
+	}
+
+	/**
+	 * Creates ParentKeys with backreference for the given path and value of the parents id.
+	 */
+	public static JdbcIdentifierBuilder forBackReferences(JdbcConverter converter, AggregatePath path,
+			@Nullable Object value) {
 
 		Identifier identifier = Identifier.of( //
-				path.getReverseColumnName(), //
+				path.getTableInfo().reverseColumnInfo().name(), //
 				value, //
 				converter.getColumnType(path.getIdDefiningParentPath().getRequiredIdProperty()) //
 		);
@@ -59,12 +71,26 @@ public class JdbcIdentifierBuilder {
 	 * @param value map key or list index qualifying the map identified by {@code path}. Must not be {@literal null}.
 	 * @return this builder. Guaranteed to be not {@literal null}.
 	 */
+	@Deprecated
 	public JdbcIdentifierBuilder withQualifier(PersistentPropertyPathExtension path, Object value) {
 
+		return withQualifier(path.getAggregatePath(), value);
+	}
+
+
+	/**
+	 * Adds a qualifier to the identifier to build. A qualifier is a map key or a list index.
+	 *
+	 * @param path path to the map that gets qualified by {@code value}. Must not be {@literal null}.
+	 * @param value map key or list index qualifying the map identified by {@code path}. Must not be {@literal null}.
+	 * @return this builder. Guaranteed to be not {@literal null}.
+	 */
+	public JdbcIdentifierBuilder withQualifier(AggregatePath path, Object value) {
+
 		Assert.notNull(path, "Path must not be null");
 		Assert.notNull(value, "Value must not be null");
 
-		identifier = identifier.withPart(path.getQualifierColumn(), value, path.getQualifierColumnType());
+		identifier = identifier.withPart(path.getTableInfo().qualifierColumnInfo().name(), value, path.getTableInfo().qualifierColumnType());
 
 		return this;
 	}

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/JdbcPropertyValueProvider.java

@@ -16,9 +16,8 @@
 package org.springframework.data.jdbc.core.convert;
 
 import org.springframework.data.mapping.model.PropertyValueProvider;
-import org.springframework.data.relational.core.mapping.PersistentPropertyPathExtension;
+import org.springframework.data.relational.core.mapping.AggregatePath;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
-import org.springframework.data.relational.core.sql.IdentifierProcessing;
 
 /**
  * {@link PropertyValueProvider} obtaining values from a {@link ResultSetAccessor}.
@@ -29,15 +28,14 @@ import org.springframework.data.relational.core.sql.IdentifierProcessing;
  */
 class JdbcPropertyValueProvider implements PropertyValueProvider<RelationalPersistentProperty> {
 
-	private final PersistentPropertyPathExtension basePath;
+	private final AggregatePath basePath;
 	private final ResultSetAccessor resultSet;
 
 	/**
 	 * @param basePath path from the aggregate root relative to which all properties get resolved.
 	 * @param resultSet the {@link ResultSetAccessor} from which to obtain the actual values.
 	 */
-	JdbcPropertyValueProvider(PersistentPropertyPathExtension basePath,
-			ResultSetAccessor resultSet) {
+	JdbcPropertyValueProvider(AggregatePath basePath, ResultSetAccessor resultSet) {
 
 		this.resultSet = resultSet;
 		this.basePath = basePath;
@@ -59,10 +57,11 @@ class JdbcPropertyValueProvider implements PropertyValueProvider<RelationalPersi
 	}
 
 	private String getColumnName(RelationalPersistentProperty property) {
-		return basePath.extendBy(property).getColumnAlias().getReference();
+		AggregatePath.ColumnInfo columnInfo = basePath.append(property).getColumnInfo();
+		return columnInfo.alias().getReference();
 	}
 
 	public JdbcPropertyValueProvider extendBy(RelationalPersistentProperty property) {
-		return new JdbcPropertyValueProvider(basePath.extendBy(property), resultSet);
+		return new JdbcPropertyValueProvider(basePath.append(property), resultSet);
 	}
 }

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/MapEntityRowMapper.java

@@ -20,6 +20,7 @@ import java.sql.SQLException;
 import java.util.HashMap;
 import java.util.Map;
 
+import org.springframework.data.relational.core.mapping.AggregatePath;
 import org.springframework.data.relational.core.mapping.PersistentPropertyPathExtension;
 import org.springframework.data.relational.core.sql.IdentifierProcessing;
 import org.springframework.data.relational.core.sql.SqlIdentifier;
@@ -35,12 +36,13 @@ import org.springframework.jdbc.core.RowMapper;
  */
 class MapEntityRowMapper<T> implements RowMapper<Map.Entry<Object, T>> {
 
-	private final PersistentPropertyPathExtension path;
+	private final AggregatePath path;
 	private final JdbcConverter converter;
 	private final Identifier identifier;
 	private final SqlIdentifier keyColumn;
 
-	MapEntityRowMapper(PersistentPropertyPathExtension path, JdbcConverter converter, Identifier identifier, SqlIdentifier keyColumn) {
+	MapEntityRowMapper(AggregatePath path, JdbcConverter converter, Identifier identifier, SqlIdentifier keyColumn) {
+
 		this.path = path;
 		this.converter = converter;
 		this.identifier = identifier;

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/SqlContext.java

@@ -15,7 +15,7 @@
  */
 package org.springframework.data.jdbc.core.convert;
 
-import org.springframework.data.relational.core.mapping.PersistentPropertyPathExtension;
+import org.springframework.data.relational.core.mapping.AggregatePath;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.data.relational.core.sql.Column;
 import org.springframework.data.relational.core.sql.SqlIdentifier;
@@ -52,18 +52,20 @@ class SqlContext {
 		return table;
 	}
 
-	Table getTable(PersistentPropertyPathExtension path) {
+	Table getTable(AggregatePath path) {
 
-		SqlIdentifier tableAlias = path.getTableAlias();
-		Table table = Table.create(path.getQualifiedTableName());
+		SqlIdentifier tableAlias = path.getTableInfo().tableAlias();
+		Table table = Table.create(path.getTableInfo().qualifiedTableName());
 		return tableAlias == null ? table : table.as(tableAlias);
 	}
 
-	Column getColumn(PersistentPropertyPathExtension path) {
-		return getTable(path).column(path.getColumnName()).as(path.getColumnAlias());
+	Column getColumn(AggregatePath path) {
+		AggregatePath.ColumnInfo columnInfo = path.getColumnInfo();
+		AggregatePath.ColumnInfo columnInfo1 = path.getColumnInfo();
+		return getTable(path).column(columnInfo1.name()).as(columnInfo.alias());
 	}
 
-	Column getReverseColumn(PersistentPropertyPathExtension path) {
-		return getTable(path).column(path.getReverseColumnName()).as(path.getReverseColumnNameAlias());
+	Column getReverseColumn(AggregatePath path) {
+		return getTable(path).column(path.getTableInfo().reverseColumnInfo().name()).as(path.getTableInfo().reverseColumnInfo().alias());
 	}
 }

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/SqlGenerator.java

@@ -26,7 +26,7 @@ import org.springframework.data.mapping.PersistentPropertyPath;
 import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.relational.core.dialect.Dialect;
 import org.springframework.data.relational.core.dialect.RenderContextFactory;
-import org.springframework.data.relational.core.mapping.PersistentPropertyPathExtension;
+import org.springframework.data.relational.core.mapping.AggregatePath;
 import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
@@ -66,8 +66,13 @@ class SqlGenerator {
 	static final SqlIdentifier IDS_SQL_PARAMETER = SqlIdentifier.unquoted("ids");
 	static final SqlIdentifier ROOT_ID_PARAMETER = SqlIdentifier.unquoted("rootId");
 
+	/**
+	 * Length of an aggregate path that is one longer then the root path.
+	 */
+	private static final int FIRST_NON_ROOT_LENTH = 2;
+
 	private final RelationalPersistentEntity<?> entity;
-	private final MappingContext<RelationalPersistentEntity<?>, RelationalPersistentProperty> mappingContext;
+	private final RelationalMappingContext mappingContext;
 	private final RenderContext renderContext;
 
 	private final SqlContext sqlContext;
@@ -112,6 +117,40 @@ class SqlGenerator {
 		this.dialect = dialect;
 	}
 
+	/**
+	 * When deleting entities there is a fundamental difference between deleting
+	 * <ol>
+	 * <li>the aggregate root.</li>
+	 * <li>a first level entity which still references the root id directly</li>
+	 * <li>and all other entities which have to use a subselect to navigate from the id of the aggregate root to something
+	 * referenced by the table in question.</li>
+	 * </ol>
+	 * For paths of the second kind this method returns {@literal true}.
+	 *
+	 * @param path the path to analyze.
+	 * @return If the given path is considered deeply nested.
+	 */
+	private static boolean isFirstNonRoot(AggregatePath path) {
+		return path.getLength() == FIRST_NON_ROOT_LENTH;
+	}
+
+	/**
+	 * When deleting entities there is a fundamental difference between deleting
+	 * <ol>
+	 * <li>the aggregate root.</li>
+	 * <li>a first level entity which still references the root id directly</li>
+	 * <li>and all other entities which have to use a subselect to navigate from the id of the aggregate root to something
+	 * referenced by the table in question.</li>
+	 * </ol>
+	 * For paths of the third kind this method returns {@literal true}.
+	 *
+	 * @param path the path to analyze.
+	 * @return If the given path is considered deeply nested.
+	 */
+	private static boolean isDeeplyNested(AggregatePath path) {
+		return path.getLength() > FIRST_NON_ROOT_LENTH;
+	}
+
 	/**
 	 * Construct an IN-condition based on a {@link Select Sub-Select} which selects the ids (or stand-ins for ids) of the
 	 * given {@literal path} to those that reference the root entities specified by the {@literal rootCondition}.
@@ -121,25 +160,25 @@ class SqlGenerator {
 	 * @param filterColumn the column to apply the IN-condition to.
 	 * @return the IN condition
 	 */
-	private Condition getSubselectCondition(PersistentPropertyPathExtension path,
-			Function<Column, Condition> rootCondition, Column filterColumn) {
+	private Condition getSubselectCondition(AggregatePath path, Function<Column, Condition> rootCondition,
+			Column filterColumn) {
 
-		PersistentPropertyPathExtension parentPath = path.getParentPath();
+		AggregatePath parentPath = path.getParentPath();
 
 		if (!parentPath.hasIdProperty()) {
-			if (parentPath.getLength() > 1) {
+			if (isDeeplyNested(parentPath)) {
 				return getSubselectCondition(parentPath, rootCondition, filterColumn);
 			}
 			return rootCondition.apply(filterColumn);
 		}
 
-		Table subSelectTable = Table.create(parentPath.getQualifiedTableName());
-		Column idColumn = subSelectTable.column(parentPath.getIdColumnName());
-		Column selectFilterColumn = subSelectTable.column(parentPath.getEffectiveIdColumnName());
+		Table subSelectTable = Table.create(parentPath.getTableInfo().qualifiedTableName());
+		Column idColumn = subSelectTable.column(parentPath.getTableInfo().idColumnName());
+		Column selectFilterColumn = subSelectTable.column(parentPath.getTableInfo().effectiveIdColumnName());
 
 		Condition innerCondition;
 
-		if (parentPath.getLength() == 1) { // if the parent is the root of the path
+		if (isFirstNonRoot(parentPath)) { // if the parent is the root of the path
 
 			// apply the rootCondition
 			innerCondition = rootCondition.apply(selectFilterColumn);
@@ -216,9 +255,9 @@ class SqlGenerator {
 		Assert.notNull(parentIdentifier, "identifier must not be null");
 		Assert.notNull(propertyPath, "propertyPath must not be null");
 
-		PersistentPropertyPathExtension path = new PersistentPropertyPathExtension(mappingContext, propertyPath);
+		AggregatePath path = mappingContext.getAggregatePath(propertyPath);
 
-		return getFindAllByProperty(parentIdentifier, path.getQualifierColumn(), path.isOrdered());
+		return getFindAllByProperty(parentIdentifier, path.getTableInfo().qualifierColumnInfo(), path.isOrdered());
 	}
 
 	/**
@@ -233,7 +272,8 @@ class SqlGenerator {
 	 *          keyColumn must not be {@code null}.
 	 * @return a SQL String.
 	 */
-	String getFindAllByProperty(Identifier parentIdentifier, @Nullable SqlIdentifier keyColumn, boolean ordered) {
+	String getFindAllByProperty(Identifier parentIdentifier, @Nullable AggregatePath.ColumnInfo keyColumn,
+			boolean ordered) {
 
 		Assert.isTrue(keyColumn != null || !ordered,
 				"If the SQL statement should be ordered a keyColumn to order by must be provided");
@@ -243,14 +283,14 @@ class SqlGenerator {
 		SelectBuilder.SelectWhere builder = selectBuilder( //
 				keyColumn == null //
 						? Collections.emptyList() //
-						: Collections.singleton(keyColumn) //
+						: Collections.singleton(keyColumn.name()) //
 		);
 
 		Condition condition = buildConditionForBackReference(parentIdentifier, table);
 		SelectBuilder.SelectWhereAndOr withWhereClause = builder.where(condition);
 
 		Select select = ordered //
-				? withWhereClause.orderBy(table.column(keyColumn).as(keyColumn)).build() //
+				? withWhereClause.orderBy(table.column(keyColumn.name()).as(keyColumn.alias())).build() //
 				: withWhereClause.build();
 
 		return render(select);
@@ -399,7 +439,7 @@ class SqlGenerator {
 			return render(deleteAll.build());
 		}
 
-		return createDeleteByPathAndCriteria(new PersistentPropertyPathExtension(mappingContext, path), Column::isNotNull);
+		return createDeleteByPathAndCriteria(mappingContext.getAggregatePath(path), Column::isNotNull);
 	}
 
 	/**
@@ -410,7 +450,7 @@ class SqlGenerator {
 	 * @return the statement as a {@link String}. Guaranteed to be not {@literal null}.
 	 */
 	String createDeleteByPath(PersistentPropertyPath<RelationalPersistentProperty> path) {
-		return createDeleteByPathAndCriteria(new PersistentPropertyPathExtension(mappingContext, path),
+		return createDeleteByPathAndCriteria(mappingContext.getAggregatePath(path),
 				filterColumn -> filterColumn.isEqualTo(getBindMarker(ROOT_ID_PARAMETER)));
 	}
 
@@ -423,7 +463,7 @@ class SqlGenerator {
 	 */
 	String createDeleteInByPath(PersistentPropertyPath<RelationalPersistentProperty> path) {
 
-		return createDeleteByPathAndCriteria(new PersistentPropertyPathExtension(mappingContext, path),
+		return createDeleteByPathAndCriteria(mappingContext.getAggregatePath(path),
 				filterColumn -> filterColumn.in(getBindMarker(IDS_SQL_PARAMETER)));
 	}
 
@@ -480,7 +520,7 @@ class SqlGenerator {
 		for (PersistentPropertyPath<RelationalPersistentProperty> path : mappingContext
 				.findPersistentPropertyPaths(entity.getType(), p -> true)) {
 
-			PersistentPropertyPathExtension extPath = new PersistentPropertyPathExtension(mappingContext, path);
+			AggregatePath extPath = mappingContext.getAggregatePath(path);
 
 			// add a join if necessary
 			Join join = getJoin(extPath);
@@ -537,13 +577,13 @@ class SqlGenerator {
 	}
 
 	/**
-	 * Create a {@link Column} for {@link PersistentPropertyPathExtension}.
+	 * Create a {@link Column} for {@link AggregatePath}.
 	 *
 	 * @param path the path to the column in question.
 	 * @return the statement as a {@link String}. Guaranteed to be not {@literal null}.
 	 */
 	@Nullable
-	Column getColumn(PersistentPropertyPathExtension path) {
+	Column getColumn(AggregatePath path) {
 
 		// an embedded itself doesn't give a column, its members will though.
 		// if there is a collection or map on the path it won't get selected at all, but it will get loaded with a separate
@@ -572,7 +612,7 @@ class SqlGenerator {
 	}
 
 	@Nullable
-	Join getJoin(PersistentPropertyPathExtension path) {
+	Join getJoin(AggregatePath path) {
 
 		if (!path.isEntity() || path.isEmbedded() || path.isMultiValued()) {
 			return null;
@@ -580,13 +620,13 @@ class SqlGenerator {
 
 		Table currentTable = sqlContext.getTable(path);
 
-		PersistentPropertyPathExtension idDefiningParentPath = path.getIdDefiningParentPath();
+		AggregatePath idDefiningParentPath = path.getIdDefiningParentPath();
 		Table parentTable = sqlContext.getTable(idDefiningParentPath);
 
 		return new Join( //
 				currentTable, //
-				currentTable.column(path.getReverseColumnName()), //
-				parentTable.column(idDefiningParentPath.getIdColumnName()) //
+				currentTable.column(path.getTableInfo().reverseColumnInfo().name()), //
+				parentTable.column(idDefiningParentPath.getTableInfo().idColumnName()) //
 		);
 	}
 
@@ -707,18 +747,17 @@ class SqlGenerator {
 				.where(getIdColumn().in(getBindMarker(IDS_SQL_PARAMETER)));
 	}
 
-	private String createDeleteByPathAndCriteria(PersistentPropertyPathExtension path,
-			Function<Column, Condition> rootCondition) {
+	private String createDeleteByPathAndCriteria(AggregatePath path, Function<Column, Condition> rootCondition) {
 
-		Table table = Table.create(path.getQualifiedTableName());
+		Table table = Table.create(path.getTableInfo().qualifiedTableName());
 
 		DeleteBuilder.DeleteWhere builder = Delete.builder() //
 				.from(table);
 		Delete delete;
 
-		Column filterColumn = table.column(path.getReverseColumnName());
+		Column filterColumn = table.column(path.getTableInfo().reverseColumnInfo().name());
 
-		if (path.getLength() == 1) {
+		if (isFirstNonRoot(path)) {
 
 			delete = builder //
 					.where(rootCondition.apply(filterColumn)) //
@@ -926,10 +965,10 @@ class SqlGenerator {
 		for (PersistentPropertyPath<RelationalPersistentProperty> path : mappingContext
 				.findPersistentPropertyPaths(entity.getType(), p -> true)) {
 
-			PersistentPropertyPathExtension extPath = new PersistentPropertyPathExtension(mappingContext, path);
+			AggregatePath aggregatePath = mappingContext.getAggregatePath(path);
 
 			// add a join if necessary
-			Join join = getJoin(extPath);
+			Join join = getJoin(aggregatePath);
 			if (join != null) {
 				baseSelect = baseSelect.leftOuterJoin(join.joinTable).on(join.joinColumn).equals(join.parentId);
 			}
@@ -960,7 +999,7 @@ class SqlGenerator {
 		for (PersistentPropertyPath<RelationalPersistentProperty> path : mappingContext
 				.findPersistentPropertyPaths(entity.getType(), p -> true)) {
 
-			PersistentPropertyPathExtension extPath = new PersistentPropertyPathExtension(mappingContext, path);
+			AggregatePath extPath = mappingContext.getAggregatePath(path);
 
 			// add a join if necessary
 			Join join = getJoin(extPath);

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/repository/query/JdbcQueryCreator.java

@@ -25,10 +25,9 @@ import org.springframework.data.domain.Sort;
 import org.springframework.data.jdbc.core.convert.JdbcConverter;
 import org.springframework.data.jdbc.core.convert.QueryMapper;
 import org.springframework.data.mapping.PersistentPropertyPath;
-import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.relational.core.dialect.Dialect;
 import org.springframework.data.relational.core.dialect.RenderContextFactory;
-import org.springframework.data.relational.core.mapping.PersistentPropertyPathExtension;
+import org.springframework.data.relational.core.mapping.AggregatePath;
 import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
@@ -117,8 +116,7 @@ class JdbcQueryCreator extends RelationalQueryCreator<ParametrizedQuery> {
 	 * @param tree the tree structure defining the predicate of the query.
 	 * @param parameters parameters for the predicate.
 	 */
-	static void validate(PartTree tree, Parameters<?, ?> parameters,
-			MappingContext<? extends RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty> context) {
+	static void validate(PartTree tree, Parameters<?, ?> parameters, RelationalMappingContext context) {
 
 		RelationalQueryCreator.validate(tree, parameters);
 
@@ -127,30 +125,30 @@ class JdbcQueryCreator extends RelationalQueryCreator<ParametrizedQuery> {
 
 				PersistentPropertyPath<? extends RelationalPersistentProperty> propertyPath = context
 						.getPersistentPropertyPath(part.getProperty());
-				PersistentPropertyPathExtension path = new PersistentPropertyPathExtension(context, propertyPath);
+				AggregatePath path = context.getAggregatePath(propertyPath);
 
-				for (PersistentPropertyPathExtension pathToValidate = path; path.getLength() > 0; path = path.getParentPath()) {
-					validateProperty(pathToValidate);
-				}
+				path.forEach(JdbcQueryCreator::validateProperty);
 			}
 		}
 	}
 
-	private static void validateProperty(PersistentPropertyPathExtension path) {
+	private static void validateProperty(AggregatePath path) {
 
-		if (!path.getParentPath().isEmbedded() && path.getLength() > 1) {
-			throw new IllegalArgumentException(
-					String.format("Cannot query by nested property: %s", path.getRequiredPersistentPropertyPath().toDotPath()));
+		if (path.isRoot()) {
+			return;
+		}
+
+		if (!path.getParentPath().isEmbedded() && path.getLength() > 2) {
+			throw new IllegalArgumentException(String.format("Cannot query by nested property: %s", path.toDotPath()));
 		}
 
 		if (path.isMultiValued() || path.isMap()) {
-			throw new IllegalArgumentException(String.format("Cannot query by multi-valued property: %s",
-					path.getRequiredPersistentPropertyPath().getLeafProperty().getName()));
+			throw new IllegalArgumentException(
+					String.format("Cannot query by multi-valued property: %s", path.getRequiredLeafProperty().getName()));
 		}
 
 		if (!path.isEmbedded() && path.isEntity()) {
-			throw new IllegalArgumentException(
-					String.format("Cannot query by nested entity: %s", path.getRequiredPersistentPropertyPath().toDotPath()));
+			throw new IllegalArgumentException(String.format("Cannot query by nested entity: %s", path.toDotPath()));
 		}
 	}
 
@@ -245,22 +243,21 @@ class JdbcQueryCreator extends RelationalQueryCreator<ParametrizedQuery> {
 		for (PersistentPropertyPath<RelationalPersistentProperty> path : context
 				.findPersistentPropertyPaths(entity.getType(), p -> true)) {
 
-			PersistentPropertyPathExtension extPath = new PersistentPropertyPathExtension(context, path);
+			AggregatePath aggregatePath = context.getAggregatePath(path);
 
 			if (returnedType.needsCustomConstruction()) {
-				if (!returnedType.getInputProperties()
-						.contains(extPath.getRequiredPersistentPropertyPath().getBaseProperty().getName())) {
+				if (!returnedType.getInputProperties().contains(aggregatePath.getRequiredBaseProperty().getName())) {
 					continue;
 				}
 			}
 
 			// add a join if necessary
-			Join join = getJoin(sqlContext, extPath);
+			Join join = getJoin(sqlContext, aggregatePath);
 			if (join != null) {
 				joinTables.add(join);
 			}
 
-			Column column = getColumn(sqlContext, extPath);
+			Column column = getColumn(sqlContext, aggregatePath);
 			if (column != null) {
 				columnExpressions.add(column);
 			}
@@ -277,14 +274,14 @@ class JdbcQueryCreator extends RelationalQueryCreator<ParametrizedQuery> {
 	}
 
 	/**
-	 * Create a {@link Column} for {@link PersistentPropertyPathExtension}.
+	 * Create a {@link Column} for {@link AggregatePath}.
 	 *
 	 * @param sqlContext
 	 * @param path the path to the column in question.
 	 * @return the statement as a {@link String}. Guaranteed to be not {@literal null}.
 	 */
 	@Nullable
-	private Column getColumn(SqlContext sqlContext, PersistentPropertyPathExtension path) {
+	private Column getColumn(SqlContext sqlContext, AggregatePath path) {
 
 		// an embedded itself doesn't give an column, its members will though.
 		// if there is a collection or map on the path it won't get selected at all, but it will get loaded with a separate
@@ -313,7 +310,7 @@ class JdbcQueryCreator extends RelationalQueryCreator<ParametrizedQuery> {
 	}
 
 	@Nullable
-	Join getJoin(SqlContext sqlContext, PersistentPropertyPathExtension path) {
+	Join getJoin(SqlContext sqlContext, AggregatePath path) {
 
 		if (!path.isEntity() || path.isEmbedded() || path.isMultiValued()) {
 			return null;
@@ -321,13 +318,13 @@ class JdbcQueryCreator extends RelationalQueryCreator<ParametrizedQuery> {
 
 		Table currentTable = sqlContext.getTable(path);
 
-		PersistentPropertyPathExtension idDefiningParentPath = path.getIdDefiningParentPath();
+		AggregatePath idDefiningParentPath = path.getIdDefiningParentPath();
 		Table parentTable = sqlContext.getTable(idDefiningParentPath);
 
 		return new Join( //
 				currentTable, //
-				currentTable.column(path.getReverseColumnName()), //
-				parentTable.column(idDefiningParentPath.getIdColumnName()) //
+				currentTable.column(path.getTableInfo().reverseColumnInfo().name()), //
+				parentTable.column(idDefiningParentPath.getTableInfo().idColumnName()) //
 		);
 	}
 

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/repository/query/SqlContext.java

@@ -15,7 +15,7 @@
  */
 package org.springframework.data.jdbc.repository.query;
 
-import org.springframework.data.relational.core.mapping.PersistentPropertyPathExtension;
+import org.springframework.data.relational.core.mapping.AggregatePath;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.data.relational.core.sql.Column;
 import org.springframework.data.relational.core.sql.SqlIdentifier;
@@ -53,18 +53,21 @@ class SqlContext {
 		return table;
 	}
 
-	Table getTable(PersistentPropertyPathExtension path) {
+	Table getTable(AggregatePath path) {
 
-		SqlIdentifier tableAlias = path.getTableAlias();
-		Table table = Table.create(path.getQualifiedTableName());
+		SqlIdentifier tableAlias = path.getTableInfo().tableAlias();
+		Table table = Table.create(path.getTableInfo().qualifiedTableName());
 		return tableAlias == null ? table : table.as(tableAlias);
 	}
 
-	Column getColumn(PersistentPropertyPathExtension path) {
-		return getTable(path).column(path.getColumnName()).as(path.getColumnAlias());
+	Column getColumn(AggregatePath path) {
+		AggregatePath.ColumnInfo columnInfo = path.getColumnInfo();
+		AggregatePath.ColumnInfo columnInfo1 = path.getColumnInfo();
+		return getTable(path).column(columnInfo1.name()).as(columnInfo.alias());
 	}
 
-	Column getReverseColumn(PersistentPropertyPathExtension path) {
-		return getTable(path).column(path.getReverseColumnName()).as(path.getReverseColumnNameAlias());
+	Column getReverseColumn(AggregatePath path) {
+		return getTable(path).column(path.getTableInfo().reverseColumnInfo().name())
+				.as(path.getTableInfo().reverseColumnInfo().alias());
 	}
 }

spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/JdbcAggregateChangeExecutorContextImmutableUnitTests.java

@@ -27,7 +27,7 @@ import org.springframework.data.mapping.PersistentPropertyPath;
 import org.springframework.data.mapping.PersistentPropertyPaths;
 import org.springframework.data.relational.core.conversion.DbAction;
 import org.springframework.data.relational.core.conversion.IdValueSource;
-import org.springframework.data.relational.core.mapping.PersistentPropertyPathExtension;
+import org.springframework.data.relational.core.mapping.AggregatePath;
 import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
 import org.springframework.lang.Nullable;
@@ -156,8 +156,8 @@ public class JdbcAggregateChangeExecutorContextImmutableUnitTests {
 				key == null ? emptyMap() : singletonMap(toPath(propertyName), key), IdValueSource.GENERATED);
 	}
 
-	PersistentPropertyPathExtension toPathExt(String path) {
-		return new PersistentPropertyPathExtension(context, getPersistentPropertyPath(path));
+	AggregatePath toAggregatePath(String path) {
+		return context.getAggregatePath(getPersistentPropertyPath(path));
 	}
 
 	PersistentPropertyPath<RelationalPersistentProperty> getPersistentPropertyPath(String propertyName) {
@@ -165,7 +165,7 @@ public class JdbcAggregateChangeExecutorContextImmutableUnitTests {
 	}
 
 	Identifier createBackRef(long value) {
-		return JdbcIdentifierBuilder.forBackReferences(converter, toPathExt("content"), value).build();
+		return JdbcIdentifierBuilder.forBackReferences(converter, toAggregatePath("content"), value).build();
 	}
 
 	PersistentPropertyPath<RelationalPersistentProperty> toPath(String path) {

spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/JdbcAggregateChangeExecutorContextUnitTests.java

@@ -34,7 +34,7 @@ import org.springframework.data.mapping.PersistentPropertyPath;
 import org.springframework.data.mapping.PersistentPropertyPaths;
 import org.springframework.data.relational.core.conversion.DbAction;
 import org.springframework.data.relational.core.conversion.IdValueSource;
-import org.springframework.data.relational.core.mapping.PersistentPropertyPathExtension;
+import org.springframework.data.relational.core.mapping.AggregatePath;
 import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
 import org.springframework.data.relational.core.sql.SqlIdentifier;
@@ -248,8 +248,8 @@ public class JdbcAggregateChangeExecutorContextUnitTests {
 				key == null ? emptyMap() : singletonMap(toPath(propertyName), key), idValueSource);
 	}
 
-	PersistentPropertyPathExtension toPathExt(String path) {
-		return new PersistentPropertyPathExtension(context, getPersistentPropertyPath(path));
+	AggregatePath toAggregatePath(String path) {
+		return context.getAggregatePath(getPersistentPropertyPath(path));
 	}
 
 	PersistentPropertyPath<RelationalPersistentProperty> getPersistentPropertyPath(String propertyName) {
@@ -257,7 +257,7 @@ public class JdbcAggregateChangeExecutorContextUnitTests {
 	}
 
 	Identifier createBackRef(long value) {
-		return forBackReferences(converter, toPathExt("content"), value).build();
+		return forBackReferences(converter, toAggregatePath("content"), value).build();
 	}
 
 	PersistentPropertyPath<RelationalPersistentProperty> toPath(String path) {

spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/PersistentPropertyPathExtensionUnitTests.java

@@ -15,6 +15,7 @@
  */
 package org.springframework.data.jdbc.core;
 
+import static org.assertj.core.api.Assertions.*;
 import static org.assertj.core.api.SoftAssertions.*;
 import static org.springframework.data.relational.core.sql.SqlIdentifier.*;
 
@@ -238,6 +239,14 @@ public class PersistentPropertyPathExtensionUnitTests {
 		});
 	}
 
+	@Test // GH-1525
+	void getAggregatePath() {
+		assertThat(extPath("withId").getAggregatePath()).isNotNull();
+	}
+	@Test // GH-1525
+	void getAggregatePathFromRoot() {
+		assertThat(extPath(entity).getAggregatePath()).isNotNull();
+	}
 	private PersistentPropertyPathExtension extPath(RelationalPersistentEntity<?> entity) {
 		return new PersistentPropertyPathExtension(context, entity);
 	}
@@ -247,7 +256,7 @@ public class PersistentPropertyPathExtensionUnitTests {
 	}
 
 	PersistentPropertyPath<RelationalPersistentProperty> createSimplePath(String path) {
-		return PropertyPathTestingUtils.toPath(path, DummyEntity.class, context);
+		return PersistentPropertyPathTestUtils.getPath(path, DummyEntity.class, context);
 	}
 
 	@SuppressWarnings("unused")

spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/PersistentPropertyPathTestUtils.java

@@ -26,18 +26,22 @@ import org.springframework.data.relational.core.mapping.RelationalPersistentProp
  *
  * @author Jens Schauder
  */
-public final class PropertyPathTestingUtils {
+public final class PersistentPropertyPathTestUtils {
 
-	private PropertyPathTestingUtils() {
+	private PersistentPropertyPathTestUtils() {
 		throw new UnsupportedOperationException("This is a utility class and cannot be instantiated");
 	}
 
-	public static PersistentPropertyPath<RelationalPersistentProperty> toPath(String path, Class source,
-																			  RelationalMappingContext context) {
+	public static PersistentPropertyPath<RelationalPersistentProperty> getPath(String path, Class source,
+																			   RelationalMappingContext context) {
 
 		PersistentPropertyPaths<?, RelationalPersistentProperty> persistentPropertyPaths = context
 				.findPersistentPropertyPaths(source, p -> true);
 
-		return persistentPropertyPaths.filter(p -> p.toDotPath().equals(path)).stream().findFirst().orElse(null);
+		return persistentPropertyPaths
+				.filter(p -> p.toDotPath().equals(path))
+				.stream()
+				.findFirst()
+				.orElse(null);
 	}
 }

spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/convert/JdbcIdentifierBuilderUnitTests.java

@@ -16,7 +16,6 @@
 package org.springframework.data.jdbc.core.convert;
 
 import static org.assertj.core.api.Assertions.*;
-import static org.springframework.data.jdbc.core.PropertyPathTestingUtils.*;
 import static org.springframework.data.relational.core.sql.SqlIdentifier.*;
 
 import java.util.List;
@@ -25,8 +24,9 @@ import java.util.UUID;
 
 import org.junit.jupiter.api.Test;
 import org.springframework.data.annotation.Id;
+import org.springframework.data.jdbc.core.PersistentPropertyPathTestUtils;
 import org.springframework.data.jdbc.core.mapping.JdbcMappingContext;
-import org.springframework.data.relational.core.mapping.PersistentPropertyPathExtension;
+import org.springframework.data.relational.core.mapping.AggregatePath;
 
 /**
  * Unit tests for the {@link JdbcIdentifierBuilder}.
@@ -55,7 +55,7 @@ public class JdbcIdentifierBuilderUnitTests {
 	@Test // DATAJDBC-326
 	public void qualifiersForMaps() {
 
-		PersistentPropertyPathExtension path = getPath("children");
+		AggregatePath path = getPath("children");
 
 		Identifier identifier = JdbcIdentifierBuilder //
 				.forBackReferences(converter, path, "parent-eins") //
@@ -73,7 +73,7 @@ public class JdbcIdentifierBuilderUnitTests {
 	@Test // DATAJDBC-326
 	public void qualifiersForLists() {
 
-		PersistentPropertyPathExtension path = getPath("moreChildren");
+		AggregatePath path = getPath("moreChildren");
 
 		Identifier identifier = JdbcIdentifierBuilder //
 				.forBackReferences(converter, path, "parent-eins") //
@@ -116,8 +116,8 @@ public class JdbcIdentifierBuilderUnitTests {
 				);
 	}
 
-	private PersistentPropertyPathExtension getPath(String dotPath) {
-		return new PersistentPropertyPathExtension(context, toPath(dotPath, DummyEntity.class, context));
+	private AggregatePath getPath(String dotPath) {
+		return context.getAggregatePath(PersistentPropertyPathTestUtils.getPath(dotPath, DummyEntity.class, context));
 	}
 
 	@SuppressWarnings("unused")

spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/convert/SqlGeneratorContextBasedNamingStrategyUnitTests.java

@@ -26,8 +26,8 @@ import org.assertj.core.api.SoftAssertions;
 
 import org.junit.jupiter.api.Test;
 import org.springframework.data.annotation.Id;
+import org.springframework.data.jdbc.core.PersistentPropertyPathTestUtils;
 import org.springframework.data.jdbc.core.mapping.JdbcMappingContext;
-import org.springframework.data.jdbc.core.mapping.PersistentPropertyPathTestUtils;
 import org.springframework.data.mapping.PersistentPropertyPath;
 import org.springframework.data.relational.core.mapping.NamingStrategy;
 import org.springframework.data.relational.core.mapping.RelationalMappingContext;
@@ -157,7 +157,7 @@ public class SqlGeneratorContextBasedNamingStrategyUnitTests {
 	}
 
 	private PersistentPropertyPath<RelationalPersistentProperty> getPath(String path) {
-		return PersistentPropertyPathTestUtils.getPath(this.context, path, DummyEntity.class);
+		return PersistentPropertyPathTestUtils.getPath(path, DummyEntity.class, this.context);
 	}
 
 	/**

spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/convert/SqlGeneratorEmbeddedUnitTests.java

@@ -23,12 +23,11 @@ import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.springframework.data.annotation.Id;
-import org.springframework.data.jdbc.core.PropertyPathTestingUtils;
+import org.springframework.data.jdbc.core.PersistentPropertyPathTestUtils;
 import org.springframework.data.jdbc.core.mapping.JdbcMappingContext;
 import org.springframework.data.relational.core.mapping.Column;
 import org.springframework.data.relational.core.mapping.Embedded;
 import org.springframework.data.relational.core.mapping.Embedded.OnEmpty;
-import org.springframework.data.relational.core.mapping.PersistentPropertyPathExtension;
 import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.data.relational.core.sql.Aliased;
@@ -177,7 +176,7 @@ public class SqlGeneratorEmbeddedUnitTests {
 	public void deleteByPath() {
 
 		final String sql = sqlGenerator
-				.createDeleteByPath(PropertyPathTestingUtils.toPath("embedded.other", DummyEntity2.class, context));
+				.createDeleteByPath(PersistentPropertyPathTestUtils.getPath("embedded.other", DummyEntity2.class, context));
 
 		assertThat(sql).containsSequence("DELETE FROM other_entity", //
 				"WHERE", //
@@ -295,7 +294,7 @@ public class SqlGeneratorEmbeddedUnitTests {
 
 	private SqlGenerator.Join generateJoin(String path, Class<?> type) {
 		return createSqlGenerator(type)
-				.getJoin(new PersistentPropertyPathExtension(context, PropertyPathTestingUtils.toPath(path, type, context)));
+				.getJoin(context.getAggregatePath(PersistentPropertyPathTestUtils.getPath(path, type, context)));
 	}
 
 	@Nullable
@@ -310,13 +309,14 @@ public class SqlGeneratorEmbeddedUnitTests {
 	private org.springframework.data.relational.core.sql.Column generatedColumn(String path, Class<?> type) {
 
 		return createSqlGenerator(type)
-				.getColumn(new PersistentPropertyPathExtension(context, PropertyPathTestingUtils.toPath(path, type, context)));
+				.getColumn(context.getAggregatePath(PersistentPropertyPathTestUtils.getPath(path, type, context)));
 	}
 
 	@SuppressWarnings("unused")
 	static class DummyEntity {
 
-		@Column("id1") @Id Long id;
+		@Column("id1")
+		@Id Long id;
 
 		@Embedded(onEmpty = OnEmpty.USE_NULL, prefix = "prefix_") CascadedEmbedded prefixedEmbeddable;
 

spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/convert/SqlGeneratorFixedNamingStrategyUnitTests.java

@@ -21,8 +21,8 @@ import org.assertj.core.api.SoftAssertions;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.data.annotation.Id;
+import org.springframework.data.jdbc.core.PersistentPropertyPathTestUtils;
 import org.springframework.data.jdbc.core.mapping.JdbcMappingContext;
-import org.springframework.data.jdbc.core.mapping.PersistentPropertyPathTestUtils;
 import org.springframework.data.mapping.PersistentPropertyPath;
 import org.springframework.data.relational.core.dialect.AnsiDialect;
 import org.springframework.data.relational.core.mapping.NamingStrategy;
@@ -70,7 +70,7 @@ class SqlGeneratorFixedNamingStrategyUnitTests {
 		}
 	};
 
-	private RelationalMappingContext context = new JdbcMappingContext();
+	private RelationalMappingContext context;
 
 	@Test // DATAJDBC-107
 	void findOneWithOverriddenFixedTableName() {
@@ -122,7 +122,7 @@ class SqlGeneratorFixedNamingStrategyUnitTests {
 		String sql = sqlGenerator.createDeleteByPath(getPath("ref"));
 
 		assertThat(sql).isEqualTo("DELETE FROM \"FIXEDCUSTOMSCHEMA\".\"FIXEDCUSTOMTABLEPREFIX_REFERENCEDENTITY\" "
-				+ "WHERE \"FIXEDCUSTOMSCHEMA\".\"FIXEDCUSTOMTABLEPREFIX_REFERENCEDENTITY\".\"DUMMY_ENTITY\" = :rootId");
+				+ "WHERE \"FIXEDCUSTOMSCHEMA\".\"FIXEDCUSTOMTABLEPREFIX_REFERENCEDENTITY\".\"FIXEDCUSTOMTABLEPREFIX_DUMMYENTITY\" = :rootId");
 	}
 
 	@Test // DATAJDBC-107
@@ -137,7 +137,7 @@ class SqlGeneratorFixedNamingStrategyUnitTests {
 						+ "WHERE \"FIXEDCUSTOMSCHEMA\".\"FIXEDCUSTOMTABLEPREFIX_SECONDLEVELREFERENCEDENTITY\".\"FIXEDCUSTOMTABLEPREFIX_REFERENCEDENTITY\" IN "
 						+ "(SELECT \"FIXEDCUSTOMSCHEMA\".\"FIXEDCUSTOMTABLEPREFIX_REFERENCEDENTITY\".\"FIXEDCUSTOMPROPERTYPREFIX_L1ID\" "
 						+ "FROM \"FIXEDCUSTOMSCHEMA\".\"FIXEDCUSTOMTABLEPREFIX_REFERENCEDENTITY\" "
-						+ "WHERE \"FIXEDCUSTOMSCHEMA\".\"FIXEDCUSTOMTABLEPREFIX_REFERENCEDENTITY\".\"DUMMY_ENTITY\" = :rootId)");
+						+ "WHERE \"FIXEDCUSTOMSCHEMA\".\"FIXEDCUSTOMTABLEPREFIX_REFERENCEDENTITY\".\"FIXEDCUSTOMTABLEPREFIX_DUMMYENTITY\" = :rootId)");
 	}
 
 	@Test // DATAJDBC-107
@@ -158,7 +158,7 @@ class SqlGeneratorFixedNamingStrategyUnitTests {
 		String sql = sqlGenerator.createDeleteAllSql(getPath("ref"));
 
 		assertThat(sql).isEqualTo("DELETE FROM \"FIXEDCUSTOMSCHEMA\".\"FIXEDCUSTOMTABLEPREFIX_REFERENCEDENTITY\" "
-				+ "WHERE \"FIXEDCUSTOMSCHEMA\".\"FIXEDCUSTOMTABLEPREFIX_REFERENCEDENTITY\".\"DUMMY_ENTITY\" IS NOT NULL");
+				+ "WHERE \"FIXEDCUSTOMSCHEMA\".\"FIXEDCUSTOMTABLEPREFIX_REFERENCEDENTITY\".\"FIXEDCUSTOMTABLEPREFIX_DUMMYENTITY\" IS NOT NULL");
 	}
 
 	@Test // DATAJDBC-107
@@ -173,7 +173,7 @@ class SqlGeneratorFixedNamingStrategyUnitTests {
 						+ "WHERE \"FIXEDCUSTOMSCHEMA\".\"FIXEDCUSTOMTABLEPREFIX_SECONDLEVELREFERENCEDENTITY\".\"FIXEDCUSTOMTABLEPREFIX_REFERENCEDENTITY\" IN "
 						+ "(SELECT \"FIXEDCUSTOMSCHEMA\".\"FIXEDCUSTOMTABLEPREFIX_REFERENCEDENTITY\".\"FIXEDCUSTOMPROPERTYPREFIX_L1ID\" "
 						+ "FROM \"FIXEDCUSTOMSCHEMA\".\"FIXEDCUSTOMTABLEPREFIX_REFERENCEDENTITY\" "
-						+ "WHERE \"FIXEDCUSTOMSCHEMA\".\"FIXEDCUSTOMTABLEPREFIX_REFERENCEDENTITY\".\"DUMMY_ENTITY\" IS NOT NULL)");
+						+ "WHERE \"FIXEDCUSTOMSCHEMA\".\"FIXEDCUSTOMTABLEPREFIX_REFERENCEDENTITY\".\"FIXEDCUSTOMTABLEPREFIX_DUMMYENTITY\" IS NOT NULL)");
 	}
 
 	@Test // DATAJDBC-113
@@ -188,7 +188,7 @@ class SqlGeneratorFixedNamingStrategyUnitTests {
 	}
 
 	private PersistentPropertyPath<RelationalPersistentProperty> getPath(String path) {
-		return PersistentPropertyPathTestUtils.getPath(context, path, DummyEntity.class);
+		return PersistentPropertyPathTestUtils.getPath(path, DummyEntity.class, context);
 	}
 
 	/**
@@ -196,7 +196,7 @@ class SqlGeneratorFixedNamingStrategyUnitTests {
 	 */
 	private SqlGenerator configureSqlGenerator(NamingStrategy namingStrategy) {
 
-		RelationalMappingContext context = new JdbcMappingContext(namingStrategy);
+		context = new JdbcMappingContext(namingStrategy);
 		JdbcConverter converter = new BasicJdbcConverter(context, (identifier, path) -> {
 			throw new UnsupportedOperationException();
 		});

spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/convert/SqlGeneratorUnitTests.java

@@ -32,18 +32,17 @@ import org.springframework.data.annotation.Version;
 import org.springframework.data.domain.PageRequest;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
-import org.springframework.data.jdbc.core.PropertyPathTestingUtils;
+import org.springframework.data.jdbc.core.PersistentPropertyPathTestUtils;
 import org.springframework.data.jdbc.core.mapping.AggregateReference;
 import org.springframework.data.jdbc.core.mapping.JdbcMappingContext;
-import org.springframework.data.jdbc.core.mapping.PersistentPropertyPathTestUtils;
 import org.springframework.data.mapping.PersistentPropertyPath;
 import org.springframework.data.relational.core.dialect.AnsiDialect;
 import org.springframework.data.relational.core.dialect.Dialect;
 import org.springframework.data.relational.core.dialect.PostgresDialect;
 import org.springframework.data.relational.core.dialect.SqlServerDialect;
+import org.springframework.data.relational.core.mapping.AggregatePath;
 import org.springframework.data.relational.core.mapping.Column;
 import org.springframework.data.relational.core.mapping.DefaultNamingStrategy;
-import org.springframework.data.relational.core.mapping.PersistentPropertyPathExtension;
 import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
@@ -389,7 +388,7 @@ class SqlGeneratorUnitTests {
 	void findAllByPropertyWithKey() {
 
 		// this would get called when ListParent is th element type of a Map
-		String sql = sqlGenerator.getFindAllByProperty(BACKREF, unquoted("key-column"), false);
+		String sql = sqlGenerator.getFindAllByProperty(BACKREF, new AggregatePath.ColumnInfo(unquoted("key-column"),unquoted("key-column")), false);
 
 		assertThat(sql).isEqualTo("SELECT dummy_entity.id1 AS id1, dummy_entity.x_name AS x_name, " //
 				+ "dummy_entity.x_other AS x_other, " //
@@ -412,7 +411,8 @@ class SqlGeneratorUnitTests {
 	void findAllByPropertyWithKeyOrdered() {
 
 		// this would get called when ListParent is th element type of a Map
-		String sql = sqlGenerator.getFindAllByProperty(BACKREF, unquoted("key-column"), true);
+		String sql = sqlGenerator.getFindAllByProperty(BACKREF,
+				new AggregatePath.ColumnInfo(unquoted("key-column"), unquoted("key-column")), true);
 
 		assertThat(sql).isEqualTo("SELECT dummy_entity.id1 AS id1, dummy_entity.x_name AS x_name, " //
 				+ "dummy_entity.x_other AS x_other, " //
@@ -432,8 +432,10 @@ class SqlGeneratorUnitTests {
 		final SqlGenerator sqlGenerator = createSqlGenerator(ReferencedEntity.class);
 		final String sql = sqlGenerator.getFindAllByProperty(
 				Identifier.of(quoted("id"), "parent-id-value", DummyEntity.class), //
-				quoted("X_L1ID"), // this key column collides with the name derived by the naming strategy for the id of
-													// ReferencedEntity.
+				new AggregatePath.ColumnInfo(quoted("X_L1ID"), quoted("X_L1ID")), // this key column collides with the name
+																																					// derived by the naming strategy for the id
+																																					// of
+				// ReferencedEntity.
 				false);
 
 		final String id = "referenced_entity.x_l1id AS x_l1id";
@@ -447,10 +449,11 @@ class SqlGeneratorUnitTests {
 	void findAllByPropertyWithEmptyBackrefColumn() {
 
 		Identifier emptyIdentifier = Identifier.of(EMPTY, 0, Object.class);
-		assertThatThrownBy(() -> sqlGenerator.getFindAllByProperty(emptyIdentifier, unquoted("key-column"), false)) //
-				.isInstanceOf(IllegalArgumentException.class) //
-				.hasMessageContaining(
-						"An empty SqlIdentifier can't be used in condition. Make sure that all composite primary keys are defined in the query");
+		assertThatThrownBy(() -> sqlGenerator.getFindAllByProperty(emptyIdentifier,
+				new AggregatePath.ColumnInfo(unquoted("key-column"), unquoted("key-column")), false)) //
+						.isInstanceOf(IllegalArgumentException.class) //
+						.hasMessageContaining(
+								"An empty SqlIdentifier can't be used in condition. Make sure that all composite primary keys are defined in the query");
 	}
 
 	@Test // DATAJDBC-219
@@ -574,15 +577,16 @@ class SqlGeneratorUnitTests {
 
 		final SqlGenerator sqlGenerator = createSqlGenerator(EntityWithReadOnlyProperty.class);
 
-		assertThat(sqlGenerator.getFindAllByProperty(BACKREF, unquoted("key-column"), true)).isEqualToIgnoringCase( //
-				"SELECT " //
-						+ "entity_with_read_only_property.x_id AS x_id, " //
-						+ "entity_with_read_only_property.x_name AS x_name, " //
-						+ "entity_with_read_only_property.x_read_only_value AS x_read_only_value, " //
-						+ "entity_with_read_only_property.key-column AS key-column " //
-						+ "FROM entity_with_read_only_property " //
-						+ "WHERE entity_with_read_only_property.backref = :backref " //
-						+ "ORDER BY key-column" //
+		assertThat(sqlGenerator.getFindAllByProperty(BACKREF,
+				new AggregatePath.ColumnInfo(unquoted("key-column"), unquoted("key-column")), true)).isEqualToIgnoringCase( //
+						"SELECT " //
+								+ "entity_with_read_only_property.x_id AS x_id, " //
+								+ "entity_with_read_only_property.x_name AS x_name, " //
+								+ "entity_with_read_only_property.x_read_only_value AS x_read_only_value, " //
+								+ "entity_with_read_only_property.key-column AS key-column " //
+								+ "FROM entity_with_read_only_property " //
+								+ "WHERE entity_with_read_only_property.backref = :backref " //
+								+ "ORDER BY key-column" //
 		);
 	}
 
@@ -736,7 +740,7 @@ class SqlGeneratorUnitTests {
 	@Nullable
 	private SqlGenerator.Join generateJoin(String path, Class<?> type) {
 		return createSqlGenerator(type, AnsiDialect.INSTANCE)
-				.getJoin(new PersistentPropertyPathExtension(context, PropertyPathTestingUtils.toPath(path, type, context)));
+				.getJoin(context.getAggregatePath(PersistentPropertyPathTestUtils.getPath(path, type, context)));
 	}
 
 	@Test // DATAJDBC-340
@@ -935,11 +939,11 @@ class SqlGeneratorUnitTests {
 	private org.springframework.data.relational.core.sql.Column generatedColumn(String path, Class<?> type) {
 
 		return createSqlGenerator(type, AnsiDialect.INSTANCE)
-				.getColumn(new PersistentPropertyPathExtension(context, PropertyPathTestingUtils.toPath(path, type, context)));
+				.getColumn(context.getAggregatePath(PersistentPropertyPathTestUtils.getPath(path, type, context)));
 	}
 
 	private PersistentPropertyPath<RelationalPersistentProperty> getPath(String path, Class<?> baseType) {
-		return PersistentPropertyPathTestUtils.getPath(context, path, baseType);
+		return PersistentPropertyPathTestUtils.getPath(path, baseType, context);
 	}
 
 	@SuppressWarnings("unused")

spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/mapping/BasicJdbcPersistentPropertyUnitTests.java

@@ -15,17 +15,11 @@
  */
 package org.springframework.data.jdbc.core.mapping;
 
+import static org.assertj.core.api.Assertions.*;
+import static org.assertj.core.api.SoftAssertions.*;
+import static org.springframework.data.relational.core.sql.SqlIdentifier.*;
+
 import junit.framework.AssertionFailedError;
-import org.junit.jupiter.api.Test;
-import org.springframework.data.annotation.Id;
-import org.springframework.data.mapping.PersistentPropertyPath;
-import org.springframework.data.relational.core.mapping.BasicRelationalPersistentProperty;
-import org.springframework.data.relational.core.mapping.Column;
-import org.springframework.data.relational.core.mapping.MappedCollection;
-import org.springframework.data.relational.core.mapping.PersistentPropertyPathExtension;
-import org.springframework.data.relational.core.mapping.RelationalMappingContext;
-import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
-import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
 
 import java.time.LocalDateTime;
 import java.time.ZonedDateTime;
@@ -33,9 +27,16 @@ import java.util.Date;
 import java.util.List;
 import java.util.UUID;
 
-import static org.assertj.core.api.Assertions.*;
-import static org.assertj.core.api.SoftAssertions.*;
-import static org.springframework.data.relational.core.sql.SqlIdentifier.*;
+import org.junit.jupiter.api.Test;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.mapping.PersistentPropertyPath;
+import org.springframework.data.relational.core.mapping.AggregatePath;
+import org.springframework.data.relational.core.mapping.BasicRelationalPersistentProperty;
+import org.springframework.data.relational.core.mapping.Column;
+import org.springframework.data.relational.core.mapping.MappedCollection;
+import org.springframework.data.relational.core.mapping.RelationalMappingContext;
+import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
+import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
 
 /**
  * Unit tests for the {@link BasicRelationalPersistentProperty}.
@@ -63,9 +64,10 @@ public class BasicJdbcPersistentPropertyUnitTests {
 
 		String propertyName = "someList";
 		RelationalPersistentProperty listProperty = entity.getRequiredPersistentProperty(propertyName);
-		PersistentPropertyPathExtension path = getPersistentPropertyPath(DummyEntity.class, propertyName);
+		AggregatePath path = getPersistentPropertyPath(DummyEntity.class, propertyName);
 
-		assertThat(listProperty.getReverseColumnName(path)).isEqualTo(quoted("dummy_column_name"));
+		assertThat(listProperty.getReverseColumnName(path.getRequiredBaseProperty().getOwner()))
+				.isEqualTo(quoted("dummy_column_name"));
 		assertThat(listProperty.getKeyColumn()).isEqualTo(quoted("dummy_key_column_name"));
 	}
 
@@ -76,10 +78,11 @@ public class BasicJdbcPersistentPropertyUnitTests {
 		RelationalPersistentProperty listProperty = context //
 				.getRequiredPersistentEntity(WithCollections.class) //
 				.getRequiredPersistentProperty(propertyName);
-		PersistentPropertyPathExtension path = getPersistentPropertyPath(DummyEntity.class, propertyName);
+		AggregatePath path = getPersistentPropertyPath(DummyEntity.class, propertyName);
 
 		assertThat(listProperty.getKeyColumn()).isEqualTo(quoted("WITH_COLLECTIONS_KEY"));
-		assertThat(listProperty.getReverseColumnName(path)).isEqualTo(quoted("some_value"));
+		assertThat(listProperty.getReverseColumnName(path.getRequiredBaseProperty().getOwner()))
+				.isEqualTo(quoted("some_value"));
 	}
 
 	@Test // DATAJDBC-331
@@ -88,10 +91,11 @@ public class BasicJdbcPersistentPropertyUnitTests {
 		RelationalPersistentProperty listProperty = context //
 				.getRequiredPersistentEntity(WithCollections.class) //
 				.getRequiredPersistentProperty("overrideList");
-		PersistentPropertyPathExtension path = getPersistentPropertyPath(WithCollections.class, "overrideList");
+		AggregatePath path = getPersistentPropertyPath(WithCollections.class, "overrideList");
 
 		assertThat(listProperty.getKeyColumn()).isEqualTo(quoted("override_key"));
-		assertThat(listProperty.getReverseColumnName(path)).isEqualTo(quoted("override_id"));
+		assertThat(listProperty.getReverseColumnName(path.getRequiredBaseProperty().getOwner()))
+				.isEqualTo(quoted("override_id"));
 	}
 
 	@Test // GH-938
@@ -126,12 +130,13 @@ public class BasicJdbcPersistentPropertyUnitTests {
 		});
 	}
 
-	private PersistentPropertyPathExtension getPersistentPropertyPath(Class<?> type, String propertyName) {
+	private AggregatePath getPersistentPropertyPath(Class<?> type, String propertyName) {
+
 		PersistentPropertyPath<RelationalPersistentProperty> path = context
 				.findPersistentPropertyPaths(type, p -> p.getName().equals(propertyName)).getFirst()
 				.orElseThrow(() -> new AssertionFailedError(String.format("Couldn't find path for '%s'", propertyName)));
 
-		return new PersistentPropertyPathExtension(context, path);
+		return context.getAggregatePath(path);
 	}
 
 	@SuppressWarnings("unused")
@@ -142,8 +147,7 @@ public class BasicJdbcPersistentPropertyUnitTests {
 	@SuppressWarnings("unused")
 	private static class DummyEntity {
 
-		@Id
-		private final Long id;
+		@Id private final Long id;
 		private final SomeEnum someEnum;
 		private final LocalDateTime localDateTime;
 		private final ZonedDateTime zonedDateTime;
@@ -152,13 +156,13 @@ public class BasicJdbcPersistentPropertyUnitTests {
 		private final UUID uuid;
 
 		@MappedCollection(idColumn = "dummy_column_name",
-				keyColumn = "dummy_key_column_name")
-		private List<Integer> someList;
+				keyColumn = "dummy_key_column_name") private List<Integer> someList;
 
 		// DATACMNS-106
 		private @Column("dummy_name") String name;
 
-		private DummyEntity(Long id, SomeEnum someEnum, LocalDateTime localDateTime, ZonedDateTime zonedDateTime, AggregateReference<DummyEntity, Long> reference, List<String> listField, UUID uuid) {
+		private DummyEntity(Long id, SomeEnum someEnum, LocalDateTime localDateTime, ZonedDateTime zonedDateTime,
+				AggregateReference<DummyEntity, Long> reference, List<String> listField, UUID uuid) {
 			this.id = id;
 			this.someEnum = someEnum;
 			this.localDateTime = localDateTime;
@@ -224,8 +228,7 @@ public class BasicJdbcPersistentPropertyUnitTests {
 
 	private static class WithCollections {
 
-		@Column(value = "some_value")
-		List<Integer> someList;
+		@Column(value = "some_value") List<Integer> someList;
 
 		@Column(value = "some_value") //
 		@MappedCollection(idColumn = "override_id", keyColumn = "override_key") //

spring-data-jdbc/src/test/java/org/springframework/data/jdbc/mapping/model/DefaultNamingStrategyUnitTests.java

@@ -58,7 +58,7 @@ public class DefaultNamingStrategyUnitTests {
 	}
 
 	@Test // DATAJDBC-184
-	public void getReverseColumnName() {
+	public void getReverseColumnInfoName() {
 		assertThat(target.getReverseColumnName(persistentEntity.getPersistentProperty("dummySubEntities")))
 				.isEqualTo("dummy_entity");
 	}

spring-data-jdbc/src/test/java/org/springframework/data/jdbc/mybatis/MyBatisDataAccessStrategyUnitTests.java

@@ -28,7 +28,7 @@ import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
 import org.springframework.data.domain.PageRequest;
 import org.springframework.data.domain.Sort;
-import org.springframework.data.jdbc.core.PropertyPathTestingUtils;
+import org.springframework.data.jdbc.core.PersistentPropertyPathTestUtils;
 import org.springframework.data.jdbc.core.convert.Identifier;
 import org.springframework.data.jdbc.core.mapping.JdbcMappingContext;
 import org.springframework.data.mapping.PersistentPropertyPath;
@@ -54,7 +54,7 @@ public class MyBatisDataAccessStrategyUnitTests {
 
 	MyBatisDataAccessStrategy accessStrategy = new MyBatisDataAccessStrategy(session, IdentifierProcessing.ANSI);
 
-	PersistentPropertyPath<RelationalPersistentProperty> path = PropertyPathTestingUtils.toPath("one.two",
+	PersistentPropertyPath<RelationalPersistentProperty> path = PersistentPropertyPathTestUtils.getPath("one.two",
 			DummyEntity.class, context);
 
 	@BeforeEach

spring-data-r2dbc/src/main/java/org/springframework/data/r2dbc/convert/MappingR2dbcConverter.java

@@ -50,6 +50,7 @@ import org.springframework.data.r2dbc.support.ArrayUtils;
 import org.springframework.data.relational.core.conversion.BasicRelationalConverter;
 import org.springframework.data.relational.core.conversion.RelationalConverter;
 import org.springframework.data.relational.core.dialect.ArrayColumns;
+import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
 import org.springframework.data.util.TypeInformation;
@@ -74,7 +75,7 @@ public class MappingR2dbcConverter extends BasicRelationalConverter implements R
 	 */
 	public MappingR2dbcConverter(
 			MappingContext<? extends RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty> context) {
-		super(context, new R2dbcCustomConversions(R2dbcCustomConversions.STORE_CONVERSIONS, Collections.emptyList()));
+		super((RelationalMappingContext) context, new R2dbcCustomConversions(R2dbcCustomConversions.STORE_CONVERSIONS, Collections.emptyList()));
 	}
 
 	/**
@@ -85,7 +86,7 @@ public class MappingR2dbcConverter extends BasicRelationalConverter implements R
 	public MappingR2dbcConverter(
 			MappingContext<? extends RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty> context,
 			CustomConversions conversions) {
-		super(context, conversions);
+		super((RelationalMappingContext) context, conversions);
 	}
 
 	// ----------------------------------

spring-data-relational/src/main/java/org/springframework/data/relational/core/conversion/BasicRelationalConverter.java

@@ -38,6 +38,7 @@ import org.springframework.data.mapping.model.ConvertingPropertyAccessor;
 import org.springframework.data.mapping.model.EntityInstantiators;
 import org.springframework.data.mapping.model.ParameterValueProvider;
 import org.springframework.data.mapping.model.SimpleTypeHolder;
+import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
 import org.springframework.data.util.TypeInformation;
@@ -61,7 +62,7 @@ import org.springframework.util.ClassUtils;
  */
 public class BasicRelationalConverter implements RelationalConverter {
 
-	private final MappingContext<? extends RelationalPersistentEntity<?>, RelationalPersistentProperty> context;
+	private final RelationalMappingContext context;
 	private final ConfigurableConversionService conversionService;
 	private final EntityInstantiators entityInstantiators;
 	private final CustomConversions conversions;
@@ -71,8 +72,7 @@ public class BasicRelationalConverter implements RelationalConverter {
 	 *
 	 * @param context must not be {@literal null}.
 	 */
-	public BasicRelationalConverter(
-			MappingContext<? extends RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty> context) {
+	public BasicRelationalConverter(RelationalMappingContext context) {
 		this(context, new CustomConversions(StoreConversions.NONE, Collections.emptyList()), new DefaultConversionService(),
 				new EntityInstantiators());
 	}
@@ -83,22 +83,18 @@ public class BasicRelationalConverter implements RelationalConverter {
 	 * @param context must not be {@literal null}.
 	 * @param conversions must not be {@literal null}.
 	 */
-	public BasicRelationalConverter(
-			MappingContext<? extends RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty> context,
-			CustomConversions conversions) {
+	public BasicRelationalConverter(RelationalMappingContext context, CustomConversions conversions) {
 		this(context, conversions, new DefaultConversionService(), new EntityInstantiators());
 	}
 
 	@SuppressWarnings("unchecked")
-	private BasicRelationalConverter(
-			MappingContext<? extends RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty> context,
-			CustomConversions conversions, ConfigurableConversionService conversionService,
-			EntityInstantiators entityInstantiators) {
+	private BasicRelationalConverter(RelationalMappingContext context, CustomConversions conversions,
+			ConfigurableConversionService conversionService, EntityInstantiators entityInstantiators) {
 
 		Assert.notNull(context, "MappingContext must not be null");
 		Assert.notNull(conversions, "CustomConversions must not be null");
 
-		this.context = (MappingContext) context;
+		this.context = context;
 		this.conversionService = conversionService;
 		this.entityInstantiators = entityInstantiators;
 		this.conversions = conversions;
@@ -116,7 +112,7 @@ public class BasicRelationalConverter implements RelationalConverter {
 	}
 
 	@Override
-	public MappingContext<? extends RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty> getMappingContext() {
+	public RelationalMappingContext getMappingContext() {
 		return context;
 	}
 

spring-data-relational/src/main/java/org/springframework/data/relational/core/conversion/RelationalEntityDeleteWriter.java

@@ -22,6 +22,7 @@ import java.util.function.Consumer;
 
 import org.springframework.data.convert.EntityWriter;
 import org.springframework.data.mapping.PersistentPropertyPath;
+import org.springframework.data.relational.core.mapping.AggregatePath;
 import org.springframework.data.relational.core.mapping.PersistentPropertyPathExtension;
 import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
@@ -126,7 +127,7 @@ public class RelationalEntityDeleteWriter implements EntityWriter<Object, Mutabl
 			Consumer<PersistentPropertyPath<RelationalPersistentProperty>> pathConsumer) {
 
 		context.findPersistentPropertyPaths(entityType, property -> property.isEntity() && !property.isEmbedded()) //
-				.filter(PersistentPropertyPathExtension::isWritable) //
+				.filter(path -> context.getAggregatePath(path).isWritable()) //
 				.forEach(pathConsumer);
 	}
 }

spring-data-relational/src/main/java/org/springframework/data/relational/core/conversion/WritingContext.java

@@ -24,6 +24,7 @@ import java.util.List;
 import java.util.Map;
 
 import org.springframework.data.mapping.PersistentPropertyPath;
+import org.springframework.data.relational.core.mapping.AggregatePath;
 import org.springframework.data.relational.core.mapping.PersistentPropertyPathExtension;
 import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
@@ -63,7 +64,7 @@ class WritingContext<T> {
 		this.rootIdValueSource = IdValueSource.forInstance(root,
 				context.getRequiredPersistentEntity(aggregateChange.getEntityType()));
 		this.paths = context.findPersistentPropertyPaths(entityType, (p) -> p.isEntity() && !p.isEmbedded()) //
-				.filter(PersistentPropertyPathExtension::isWritable).toList();
+				.filter(ppp -> context.getAggregatePath(ppp).isWritable()).toList();
 	}
 
 	/**

spring-data-relational/src/main/java/org/springframework/data/relational/core/mapping/AggregatePath.java

@@ -0,0 +1,342 @@
+/*
+ * Copyright 2023 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.relational.core.mapping;
+
+import java.util.function.Predicate;
+import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
+
+import org.springframework.data.mapping.PersistentProperty;
+import org.springframework.data.mapping.PersistentPropertyPath;
+import org.springframework.data.relational.core.sql.SqlIdentifier;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * Represents a path within an aggregate starting from the aggregate root. The path can be iterated from the leaf to its
+ * root.
+ *
+ * @since 3.2
+ * @author Jens Schauder
+ * @author Mark Paluch
+ */
+public interface AggregatePath extends Iterable<AggregatePath> {
+
+	/**
+	 * Returns the path that has the same beginning but is one segment shorter than this path.
+	 *
+	 * @return the parent path. Guaranteed to be not {@literal null}.
+	 * @throws IllegalStateException when called on an empty path.
+	 */
+	AggregatePath getParentPath();
+
+	/**
+	 * Creates a new path by extending the current path by the property passed as an argument.
+	 *
+	 * @param property must not be {@literal null}.
+	 * @return Guaranteed to be not {@literal null}.
+	 */
+	AggregatePath append(RelationalPersistentProperty property);
+
+	/**
+	 * @return {@literal true} if this is a root path for the underlying type.
+	 */
+	boolean isRoot();
+
+	/**
+	 * Returns the path length for the aggregate path.
+	 *
+	 * @return the path length for the aggregate path
+	 */
+	default int getLength() {
+		return 1 + (isRoot() ? 0 : getRequiredPersistentPropertyPath().getLength());
+	}
+
+	boolean isWritable();
+
+	/**
+	 * @return {@literal true} when this is an empty path or the path references an entity.
+	 */
+	boolean isEntity();
+
+	/**
+	 * Returns {@literal true} exactly when the path is non-empty and the leaf property an embedded one.
+	 *
+	 * @return if the leaf property is embedded.
+	 */
+	boolean isEmbedded();
+
+	/**
+	 * Returns {@literal true} if there are multiple values for this path, i.e. if the path contains at least one element
+	 * that is a collection and array or a map. // TODO: why does this return true if the parent entity is a collection?
+	 * This seems to mix some concepts that belong to somewhere else. // TODO: Multi-valued could be understood for
+	 * embeddables with more than one column (i.e. composite primary keys)
+	 *
+	 * @return {@literal true} if the path contains a multivalued element.
+	 */
+	boolean isMultiValued();
+
+	/**
+	 * @return {@literal true} when this is references a {@link java.util.List} or {@link java.util.Map}.
+	 */
+	boolean isQualified();
+
+	/**
+	 * @return {@literal true} if the leaf property of this path is a {@link java.util.Map}.
+	 * @see RelationalPersistentProperty#isMap()
+	 */
+	boolean isMap();
+
+	/**
+	 * @return {@literal true} when this is references a {@link java.util.Collection} or an array.
+	 */
+	boolean isCollectionLike();
+
+	/**
+	 * @return whether the leaf end of the path is ordered, i.e. the data to populate must be ordered.
+	 * @see RelationalPersistentProperty#isOrdered()
+	 */
+	boolean isOrdered();
+
+	/**
+	 * @return {@literal true} if this path represents an entity which has an identifier attribute.
+	 */
+	boolean hasIdProperty();
+
+	RelationalPersistentProperty getRequiredIdProperty();
+
+	/**
+	 * @return the persistent property path if the path is not a {@link #isRoot() root} path.
+	 * @throws IllegalStateException if the current path is a {@link #isRoot() root} path.
+	 * @see PersistentPropertyPath#getBaseProperty()
+	 */
+	PersistentPropertyPath<RelationalPersistentProperty> getRequiredPersistentPropertyPath();
+
+	/**
+	 * @return the base property.
+	 * @throws IllegalStateException if the current path is a {@link #isRoot() root} path.
+	 * @see PersistentPropertyPath#getBaseProperty()
+	 */
+	default RelationalPersistentProperty getRequiredBaseProperty() {
+		return getRequiredPersistentPropertyPath().getBaseProperty();
+	}
+
+	/**
+	 * @return the leaf property.
+	 * @throws IllegalStateException if the current path is a {@link #isRoot() root} path.
+	 * @see PersistentPropertyPath#getLeafProperty()
+	 */
+	default RelationalPersistentProperty getRequiredLeafProperty() {
+		return getRequiredPersistentPropertyPath().getLeafProperty();
+	}
+
+	/**
+	 * The {@link RelationalPersistentEntity} associated with the leaf of this path.
+	 *
+	 * @return Might return {@literal null} when called on a path that does not represent an entity.
+	 */
+	@Nullable
+	RelationalPersistentEntity<?> getLeafEntity();
+
+	/**
+	 * The {@link RelationalPersistentEntity} associated with the leaf of this path or throw {@link IllegalStateException}
+	 * if the leaf cannot be resolved.
+	 *
+	 * @return the required {@link RelationalPersistentEntity} associated with the leaf of this path.
+	 * @throws IllegalStateException if the persistent entity cannot be resolved.
+	 */
+	default RelationalPersistentEntity<?> getRequiredLeafEntity() {
+
+		RelationalPersistentEntity<?> entity = getLeafEntity();
+
+		if (entity == null) {
+
+			throw new IllegalStateException(String.format("Couldn't resolve leaf PersistentEntity for type %s",
+					getRequiredLeafProperty().getActualType()));
+		}
+
+		return entity;
+	}
+
+	/**
+	 * Returns the dot based path notation using {@link PersistentProperty#getName()}.
+	 *
+	 * @return will never be {@literal null}.
+	 */
+	String toDotPath();
+
+	// TODO: Conceptually, AggregatePath works with properties. The mapping into columns and tables should reside in a
+	// utility that can distinguish whether a property maps to one or many columns (e.g. embedded) and the same for
+	// identifier columns.
+	default TableInfo getTableInfo() {
+		return TableInfo.of(this);
+	}
+
+	default ColumnInfo getColumnInfo() {
+		return ColumnInfo.of(this);
+	}
+
+	/**
+	 * Filter the AggregatePath returning the first item matching the given {@link Predicate}.
+	 *
+	 * @param predicate must not be {@literal null}.
+	 * @return the first matching element or {@literal null}.
+	 */
+	@Nullable
+	default AggregatePath filter(Predicate<? super AggregatePath> predicate) {
+
+		for (AggregatePath item : this) {
+			if (predicate.test(item)) {
+				return item;
+			}
+		}
+
+		return null;
+	}
+
+	/**
+	 * Creates a non-parallel {@link Stream} of the underlying {@link Iterable}.
+	 *
+	 * @return will never be {@literal null}.
+	 */
+	default Stream<AggregatePath> stream() {
+		return StreamSupport.stream(spliterator(), false);
+	}
+
+	// path navigation
+
+	/**
+	 * Returns the longest ancestor path that has an {@link org.springframework.data.annotation.Id} property.
+	 *
+	 * @return A path that starts just as this path but is shorter. Guaranteed to be not {@literal null}. TODO: throws
+	 *         NoSuchElementException: No value present for empty paths
+	 */
+	AggregatePath getIdDefiningParentPath();
+
+	record TableInfo(
+
+			/*
+			 * The fully qualified name of the table this path is tied to or of the longest ancestor path that is actually
+			 * tied to a table.
+			 */
+			SqlIdentifier qualifiedTableName,
+
+			/*
+			 * The alias used for the table on which this path is based.
+			 */
+			@Nullable SqlIdentifier tableAlias,
+
+			ColumnInfo reverseColumnInfo,
+
+			/*
+			 * The column used for the list index or map key of the leaf property of this path.
+			 */
+			@Nullable ColumnInfo qualifierColumnInfo,
+
+			/*
+			 * The type of the qualifier column of the leaf property of this path or {@literal null} if this is not
+			 * applicable.
+			 */
+			@Nullable Class<?> qualifierColumnType,
+
+			/*
+			 * The column name of the id column of the ancestor path that represents an actual table.
+			 */
+			SqlIdentifier idColumnName,
+
+			/*
+			 * If the table owning ancestor has an id the column name of that id property is returned. Otherwise the reverse
+			 * column is returned.
+			 */
+			SqlIdentifier effectiveIdColumnName) {
+
+		static TableInfo of(AggregatePath path) {
+
+			AggregatePath tableOwner = AggregatePathTraversal.getTableOwningPath(path);
+
+			RelationalPersistentEntity<?> leafEntity = tableOwner.getRequiredLeafEntity();
+			SqlIdentifier qualifiedTableName = leafEntity.getQualifiedTableName();
+
+			SqlIdentifier tableAlias = tableOwner.isRoot() ? null : AggregatePathTableUtils.constructTableAlias(tableOwner);
+
+			ColumnInfo reverseColumnInfo = null;
+			if (!tableOwner.isRoot()) {
+
+				AggregatePath idDefiningParentPath = tableOwner.getIdDefiningParentPath();
+				RelationalPersistentProperty leafProperty = tableOwner.getRequiredLeafProperty();
+
+				SqlIdentifier reverseColumnName = leafProperty
+						.getReverseColumnName(idDefiningParentPath.getRequiredLeafEntity());
+
+				reverseColumnInfo = new ColumnInfo(reverseColumnName,
+						AggregatePathTableUtils.prefixWithTableAlias(path, reverseColumnName));
+			}
+
+			ColumnInfo qualifierColumnInfo = null;
+			if (!path.isRoot()) {
+
+				SqlIdentifier keyColumn = path.getRequiredLeafProperty().getKeyColumn();
+				if (keyColumn != null) {
+					qualifierColumnInfo = new ColumnInfo(keyColumn, keyColumn);
+				}
+			}
+
+			Class<?> qualifierColumnType = null;
+			if (!path.isRoot() && path.getRequiredLeafProperty().isQualified()) {
+				qualifierColumnType = path.getRequiredLeafProperty().getQualifierColumnType();
+			}
+
+			SqlIdentifier idColumnName = leafEntity.hasIdProperty() ? leafEntity.getIdColumn() : null;
+
+			SqlIdentifier effectiveIdColumnName = tableOwner.isRoot() ? idColumnName : reverseColumnInfo.name();
+
+			return new TableInfo(qualifiedTableName, tableAlias, reverseColumnInfo, qualifierColumnInfo, qualifierColumnType,
+					idColumnName, effectiveIdColumnName);
+
+		}
+	}
+
+	record ColumnInfo(
+
+			/* The name of the column used to represent this property in the database. */
+			SqlIdentifier name, /* The alias for the column used to represent this property in the database. */
+			SqlIdentifier alias) {
+
+		/**
+		 * Create a {@link ColumnInfo} from an aggregate path. ColumnInfo can be created for simple type single-value
+		 * properties only.
+		 *
+		 * @param path
+		 * @return the {@link ColumnInfo}.
+		 * @throws IllegalArgumentException if the path is {@link #isRoot()}, {@link #isEmbedded()} or
+		 *           {@link #isMultiValued()}.
+		 */
+		static ColumnInfo of(AggregatePath path) {
+
+			Assert.notNull(path, "AggregatePath must not be null");
+			Assert.isTrue(!path.isRoot(), () -> "Cannot obtain ColumnInfo for root path");
+			Assert.isTrue(!path.isEmbedded(), () -> "Cannot obtain ColumnInfo for embedded path");
+
+			// TODO: Multi-valued paths cannot be represented with a single column
+			// Assert.isTrue(!path.isMultiValued(), () -> "Cannot obtain ColumnInfo for multi-valued path");
+
+			SqlIdentifier name = AggregatePathTableUtils.assembleColumnName(path,
+					path.getRequiredLeafProperty().getColumnName());
+			return new ColumnInfo(name, AggregatePathTableUtils.prefixWithTableAlias(path, name));
+		}
+	}
+}

spring-data-relational/src/main/java/org/springframework/data/relational/core/mapping/AggregatePathTableUtils.java

@@ -0,0 +1,94 @@
+/*
+ * Copyright 2023 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.relational.core.mapping;
+
+import java.util.Collections;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.function.BinaryOperator;
+import java.util.function.Function;
+import java.util.function.Supplier;
+import java.util.stream.Collector;
+
+import org.springframework.data.relational.core.sql.SqlIdentifier;
+
+/**
+ * Utility methods to derive table and column information from {@link AggregatePath}.
+ *
+ * @author Mark Paluch
+ * @since 3.2
+ */
+class AggregatePathTableUtils {
+
+	public static SqlIdentifier assembleColumnName(AggregatePath path, SqlIdentifier suffix) {
+		return suffix.transform(constructEmbeddedPrefix(path)::concat);
+	}
+
+	private static String constructEmbeddedPrefix(AggregatePath path) {
+
+		return path.stream() //
+				.filter(p -> p != path) //
+				.takeWhile(AggregatePath::isEmbedded).map(p -> p.getRequiredLeafProperty().getEmbeddedPrefix()) //
+				.collect(new ReverseJoinCollector());
+	}
+
+	public static SqlIdentifier prefixWithTableAlias(AggregatePath path, SqlIdentifier columnName) {
+
+		AggregatePath tableOwner = AggregatePathTraversal.getTableOwningPath(path);
+		SqlIdentifier tableAlias = tableOwner.isRoot() ? null : constructTableAlias(tableOwner);
+
+		return tableAlias == null ? columnName : columnName.transform(name -> tableAlias.getReference() + "_" + name);
+	}
+
+	public static SqlIdentifier constructTableAlias(AggregatePath path) {
+
+		String alias = path.stream() //
+				.filter(p -> !p.isRoot()) //
+				.map(p -> p.isEmbedded() //
+						? p.getRequiredLeafProperty().getEmbeddedPrefix()//
+						: p.getRequiredLeafProperty().getName() + (p == path ? "" : "_") //
+				) //
+				.collect(new ReverseJoinCollector());
+		return SqlIdentifier.quoted(alias);
+	}
+
+	private static class ReverseJoinCollector implements Collector<String, StringBuilder, String> {
+		@Override
+		public Supplier<StringBuilder> supplier() {
+			return StringBuilder::new;
+		}
+
+		@Override
+		public BiConsumer<StringBuilder, String> accumulator() {
+			return ((stringBuilder, s) -> stringBuilder.insert(0, s));
+		}
+
+		@Override
+		public BinaryOperator<StringBuilder> combiner() {
+			return (a, b) -> b.append(a);
+		}
+
+		@Override
+		public Function<StringBuilder, String> finisher() {
+			return StringBuilder::toString;
+		}
+
+		@Override
+		public Set<Characteristics> characteristics() {
+			return Collections.emptySet();
+		}
+	}
+}

spring-data-relational/src/main/java/org/springframework/data/relational/core/mapping/AggregatePathTraversal.java

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2023 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.relational.core.mapping;
+
+import java.util.NoSuchElementException;
+import java.util.function.Predicate;
+
+/**
+ * @author Mark Paluch
+ */
+class AggregatePathTraversal {
+
+	public static AggregatePath getIdDefiningPath(AggregatePath aggregatePath) {
+
+		Predicate<AggregatePath> idDefiningPathFilter = ap -> !ap.equals(aggregatePath)
+				&& (ap.isRoot() || ap.hasIdProperty());
+
+		AggregatePath result = aggregatePath.filter(idDefiningPathFilter);
+		if (result == null) {
+			throw new NoSuchElementException();
+		}
+		return result;
+	}
+
+	public static AggregatePath getTableOwningPath(AggregatePath aggregatePath) {
+
+		Predicate<AggregatePath> idDefiningPathFilter = ap -> ap.isEntity() && !ap.isEmbedded();
+
+		AggregatePath result = aggregatePath.filter(idDefiningPathFilter);
+		if (result == null) {
+			throw new NoSuchElementException();
+		}
+		return result;
+	}
+
+}

spring-data-relational/src/main/java/org/springframework/data/relational/core/mapping/BasicRelationalPersistentProperty.java

@@ -225,6 +225,13 @@ public class BasicRelationalPersistentProperty extends AnnotationBasedPersistent
 		return createSqlIdentifier(expressionEvaluator.evaluate(collectionIdColumnNameExpression));
 	}
 
+	@Override
+	public SqlIdentifier getReverseColumnName(RelationalPersistentEntity<?> owner) {
+
+		return collectionIdColumnName.get()
+				.orElseGet(() -> createDerivedSqlIdentifier(this.namingStrategy.getReverseColumnName(owner)));
+	}
+
 	@Override
 	public SqlIdentifier getKeyColumn() {
 

spring-data-relational/src/main/java/org/springframework/data/relational/core/mapping/CachingNamingStrategy.java

@@ -72,6 +72,11 @@ class CachingNamingStrategy implements NamingStrategy {
 		return delegate.getReverseColumnName(property);
 	}
 
+	@Override
+	public String getReverseColumnName(RelationalPersistentEntity<?> owner) {
+		return delegate.getReverseColumnName(owner);
+	}
+
 	@Override
 	public String getSchema() {
 		return schema.get();

spring-data-relational/src/main/java/org/springframework/data/relational/core/mapping/DefaultAggregatePath.java

@@ -0,0 +1,255 @@
+/*
+ * Copyright 2023 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.relational.core.mapping;
+
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+import java.util.Objects;
+
+import org.springframework.data.mapping.PersistentPropertyPath;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * Represents a path within an aggregate starting from the aggregate root.
+ *
+ * @since 3.2
+ * @author Jens Schauder
+ * @author Mark Paluch
+ */
+class DefaultAggregatePath implements AggregatePath {
+
+	private final RelationalMappingContext context;
+
+	private final @Nullable RelationalPersistentEntity<?> rootType;
+
+	private final @Nullable PersistentPropertyPath<RelationalPersistentProperty> path;
+
+	@SuppressWarnings("unchecked")
+	DefaultAggregatePath(RelationalMappingContext context,
+			PersistentPropertyPath<? extends RelationalPersistentProperty> path) {
+
+		Assert.notNull(context, "context must not be null");
+		Assert.notNull(path, "path must not be null");
+
+		this.context = context;
+		this.path = (PersistentPropertyPath) path;
+		this.rootType = null;
+	}
+
+	DefaultAggregatePath(RelationalMappingContext context, RelationalPersistentEntity<?> rootType) {
+
+		Assert.notNull(context, "context must not be null");
+		Assert.notNull(rootType, "rootType must not be null");
+
+		this.context = context;
+		this.rootType = rootType;
+		this.path = null;
+	}
+
+	/**
+	 * Returns the path that has the same beginning but is one segment shorter than this path.
+	 *
+	 * @return the parent path. Guaranteed to be not {@literal null}.
+	 * @throws IllegalStateException when called on an empty path.
+	 */
+	@Override
+	public AggregatePath getParentPath() {
+
+		if (isRoot()) {
+			throw new IllegalStateException("The parent path of a root path is not defined.");
+		}
+
+		PersistentPropertyPath<RelationalPersistentProperty> path = getRequiredPersistentPropertyPath();
+
+		if (path.getLength() == 1) {
+			return context.getAggregatePath(path.getLeafProperty().getOwner());
+		}
+
+		return context.getAggregatePath(path.getParentPath());
+	}
+
+	@Override
+	public AggregatePath append(RelationalPersistentProperty property) {
+
+		PersistentPropertyPath<? extends RelationalPersistentProperty> newPath = isRoot() //
+				? context.getPersistentPropertyPath(property.getName(), rootType.getType()) //
+				: context.getPersistentPropertyPath(path.toDotPath() + "." + property.getName(),
+						path.getBaseProperty().getOwner().getType());
+
+		return context.getAggregatePath(newPath);
+	}
+
+	@Override
+	public boolean isRoot() {
+		return path == null;
+	}
+
+	@Override
+	public boolean isWritable() {
+		return stream().allMatch(path -> path.isRoot() || path.getRequiredLeafProperty().isWritable());
+	}
+
+	@Override
+	public boolean isEntity() {
+		return isRoot() || getRequiredLeafProperty().isEntity();
+	}
+
+	@Override
+	public boolean isEmbedded() {
+		return !isRoot() && getRequiredLeafProperty().isEmbedded();
+	}
+
+	@Override
+	public boolean isMultiValued() {
+
+		return !isRoot() && //
+				(getRequiredLeafProperty().isCollectionLike() //
+						|| getRequiredLeafProperty().isQualified() //
+						// TODO: Considering the parent as multi-valued burries the scope of this method.
+						// this needs to be resolved
+						|| getParentPath().isMultiValued() //
+				);
+	}
+
+	@Override
+	public boolean isQualified() {
+		return !isRoot() && getRequiredLeafProperty().isQualified();
+	}
+
+	@Override
+	public boolean isMap() {
+		return !isRoot() && getRequiredLeafProperty().isMap();
+	}
+
+	@Override
+	public boolean isCollectionLike() {
+		return !isRoot() && getRequiredLeafProperty().isCollectionLike();
+	}
+
+	@Override
+	public boolean isOrdered() {
+		return !isRoot() && getRequiredLeafProperty().isOrdered();
+	}
+
+	@Override
+	public boolean hasIdProperty() {
+
+		RelationalPersistentEntity<?> leafEntity = getLeafEntity();
+		return leafEntity != null && leafEntity.hasIdProperty();
+	}
+
+	@Override
+	public RelationalPersistentProperty getRequiredIdProperty() {
+		return isRoot() ? rootType.getRequiredIdProperty() : getRequiredLeafEntity().getRequiredIdProperty();
+	}
+
+	@Override
+	public PersistentPropertyPath<RelationalPersistentProperty> getRequiredPersistentPropertyPath() {
+
+		Assert.state(path != null, "Root Aggregate Paths are not associated with a PersistentPropertyPath");
+		return path;
+	}
+
+	@Override
+	public RelationalPersistentEntity<?> getLeafEntity() {
+		return isRoot() ? rootType : context.getPersistentEntity(getRequiredLeafProperty().getActualType());
+	}
+
+	@Override
+	public String toDotPath() {
+		return isRoot() ? "" : getRequiredPersistentPropertyPath().toDotPath();
+	}
+
+	@Override
+	public AggregatePath getIdDefiningParentPath() {
+		return AggregatePathTraversal.getIdDefiningPath(this);
+	}
+
+	/**
+	 * Finds and returns the longest path with ich identical or an ancestor to the current path and maps directly to a
+	 * table.
+	 *
+	 * @return a path. Guaranteed to be not {@literal null}.
+	 */
+	private AggregatePath getTableOwningAncestor() {
+		return AggregatePathTraversal.getTableOwningPath(this);
+	}
+
+	@Override
+	public String toString() {
+		return "AggregatePath["
+				+ (rootType == null ? path.getBaseProperty().getOwner().getType().getName() : rootType.getName()) + "]"
+				+ ((isRoot()) ? "/" : path.toDotPath());
+	}
+
+
+
+	@Override
+	public boolean equals(Object o) {
+
+		if (this == o)
+			return true;
+		if (o == null || getClass() != o.getClass())
+			return false;
+		DefaultAggregatePath that = (DefaultAggregatePath) o;
+		return Objects.equals(context, that.context) && Objects.equals(rootType, that.rootType)
+				&& Objects.equals(path, that.path);
+	}
+
+	@Override
+	public int hashCode() {
+		return Objects.hash(context, rootType, path);
+	}
+
+	/**
+	 * Creates an {@link Iterator} that iterates over the current path and all ancestors. It will start with the current
+	 * path, followed by its parent until ending with the root.
+	 */
+	@Override
+	public Iterator<AggregatePath> iterator() {
+		return new AggregatePathIterator(this);
+	}
+
+	private static class AggregatePathIterator implements Iterator<AggregatePath> {
+
+		private @Nullable AggregatePath current;
+
+		public AggregatePathIterator(AggregatePath current) {
+			this.current = current;
+		}
+
+		@Override
+		public boolean hasNext() {
+			return current != null;
+		}
+
+		@Override
+		public AggregatePath next() {
+
+			AggregatePath element = current;
+
+			if (element == null) {
+				throw new NoSuchElementException();
+			}
+
+			current = element.isRoot() ? null : element.getParentPath();
+
+			return element;
+		}
+	}
+
+}

spring-data-relational/src/main/java/org/springframework/data/relational/core/mapping/DefaultNamingStrategy.java

@@ -61,12 +61,17 @@ public class DefaultNamingStrategy implements NamingStrategy {
 		return getColumnNameReferencing(leafEntity);
 	}
 
-	private String getColumnNameReferencing(RelationalPersistentEntity<?> leafEntity) {
+	@Override
+	public String getReverseColumnName(RelationalPersistentEntity<?> parent) {
+		return getColumnNameReferencing(parent);
+	}
+
+	private String getColumnNameReferencing(RelationalPersistentEntity<?> entity) {
 
 		if (foreignKeyNaming == ForeignKeyNaming.IGNORE_RENAMING) {
-			return getTableName(leafEntity.getType());
+			return getTableName(entity.getType());
 		}
 
-		return leafEntity.getTableName().getReference();
+		return entity.getTableName().getReference();
 	}
 }

spring-data-relational/src/main/java/org/springframework/data/relational/core/mapping/NamingStrategy.java

@@ -15,7 +15,6 @@
  */
 package org.springframework.data.relational.core.mapping;
 
-import org.springframework.data.relational.core.sql.IdentifierProcessing;
 import org.springframework.data.util.ParsingUtils;
 import org.springframework.util.Assert;
 
@@ -88,8 +87,16 @@ public interface NamingStrategy {
 		return property.getOwner().getTableName().getReference();
 	}
 
+	/**
+	 * @deprecated use {@link #getReverseColumnName(AggregatePath)} instead.
+	 */
+	@Deprecated(since = "3.2", forRemoval = true)
 	default String getReverseColumnName(PersistentPropertyPathExtension path) {
-		return getTableName(path.getIdDefiningParentPath().getRequiredLeafEntity().getType());
+		return getReverseColumnName(path.getRequiredLeafEntity());
+	}
+
+	default String getReverseColumnName(RelationalPersistentEntity<?> owner) {
+		return getTableName(owner.getType());
 	}
 
 	/**
@@ -104,4 +111,5 @@ public interface NamingStrategy {
 
 		return getReverseColumnName(property) + "_key";
 	}
+
 }

spring-data-relational/src/main/java/org/springframework/data/relational/core/mapping/PersistentPropertyPathExtension.java

@@ -20,7 +20,6 @@ import java.util.Objects;
 import org.springframework.data.mapping.PersistentProperty;
 import org.springframework.data.mapping.PersistentPropertyPath;
 import org.springframework.data.mapping.context.MappingContext;
-import org.springframework.data.relational.core.sql.IdentifierProcessing;
 import org.springframework.data.relational.core.sql.SqlIdentifier;
 import org.springframework.data.util.Lazy;
 import org.springframework.lang.Nullable;
@@ -35,7 +34,9 @@ import org.springframework.util.StringUtils;
  * @author Daniil Razorenov
  * @author Kurt Niemi
  * @since 1.1
+ * @deprecated use {@link AggregatePath} instead
  */
+@Deprecated(since = "3.2", forRemoval = true)
 public class PersistentPropertyPathExtension {
 
 	private final RelationalPersistentEntity<?> entity;
@@ -86,7 +87,7 @@ public class PersistentPropertyPathExtension {
 	}
 
 	/**
-	 * Returns {@literal true} exactly when the path is non empty and the leaf property an embedded one.
+	 * Returns {@literal true} exactly when the path is non-empty and the leaf property an embedded one.
 	 *
 	 * @return if the leaf property is embedded.
 	 */
@@ -155,8 +156,8 @@ public class PersistentPropertyPathExtension {
 			if (this.path == null) {
 				throw new IllegalStateException("Couldn't resolve leaf PersistentEntity absent path");
 			}
-			throw new IllegalStateException(String.format("Couldn't resolve leaf PersistentEntity for type %s",
-					path.getLeafProperty().getActualType()));
+			throw new IllegalStateException(
+					String.format("Couldn't resolve leaf PersistentEntity for type %s", path.getLeafProperty().getActualType()));
 		}
 
 		return entity;
@@ -387,14 +388,6 @@ public class PersistentPropertyPathExtension {
 				: path.getLeafProperty().getActualType();
 	}
 
-	/**
-	 * @return whether the leaf end of the path is ordered, i.e. the data to populate must be ordered.
-	 * @see RelationalPersistentProperty#isOrdered()
-	 */
-	public boolean isOrdered() {
-		return path != null && path.getLeafProperty().isOrdered();
-	}
-
 	/**
 	 * @return {@literal true} if the leaf property of this path is a {@link java.util.Map}.
 	 * @see RelationalPersistentProperty#isMap()
@@ -481,8 +474,7 @@ public class PersistentPropertyPathExtension {
 	private SqlIdentifier prefixWithTableAlias(SqlIdentifier columnName) {
 
 		SqlIdentifier tableAlias = getTableAlias();
-		return tableAlias == null ? columnName
-				: columnName.transform(name -> tableAlias.getReference() + "_" + name);
+		return tableAlias == null ? columnName : columnName.transform(name -> tableAlias.getReference() + "_" + name);
 	}
 
 	@Override
@@ -500,4 +492,13 @@ public class PersistentPropertyPathExtension {
 	public int hashCode() {
 		return Objects.hash(entity, path);
 	}
+
+	public AggregatePath getAggregatePath() {
+		if (path != null) {
+
+			return ((RelationalMappingContext) context).getAggregatePath(path);
+		} else {
+			return ((RelationalMappingContext) context).getAggregatePath(entity);
+		}
+	}
 }

spring-data-relational/src/main/java/org/springframework/data/relational/core/mapping/RelationalMappingContext.java

@@ -15,8 +15,12 @@
  */
 package org.springframework.data.relational.core.mapping;
 
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
 import org.springframework.beans.BeansException;
 import org.springframework.context.ApplicationContext;
+import org.springframework.data.mapping.PersistentPropertyPath;
 import org.springframework.data.mapping.context.AbstractMappingContext;
 import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.mapping.model.Property;
@@ -39,6 +43,8 @@ public class RelationalMappingContext
 		extends AbstractMappingContext<RelationalPersistentEntity<?>, RelationalPersistentProperty> {
 
 	private final NamingStrategy namingStrategy;
+	private final Map<Object, AggregatePath> aggregatePathCache = new ConcurrentHashMap<>();
+
 	private boolean forceQuote = true;
 
 	private final ExpressionEvaluator expressionEvaluator = new ExpressionEvaluator(EvaluationContextProvider.DEFAULT);
@@ -125,8 +131,39 @@ public class RelationalMappingContext
 	}
 
 	protected void applyDefaults(BasicRelationalPersistentProperty persistentProperty) {
+
 		persistentProperty.setForceQuote(isForceQuote());
 		persistentProperty.setExpressionEvaluator(this.expressionEvaluator);
 	}
 
+	/**
+	 * Provides an {@link AggregatePath} for the provided {@link PersistentPropertyPath}.
+	 *
+	 * @param path the path to provide an {@link AggregatePath} for. Must not be null.
+	 * @return an {@link AggregatePath} on the provided path.
+	 * @since 3.2
+	 */
+	public AggregatePath getAggregatePath(PersistentPropertyPath<? extends RelationalPersistentProperty> path) {
+
+		AggregatePath aggregatePath = aggregatePathCache.get(path);
+		if (aggregatePath == null) {
+
+			aggregatePath = new DefaultAggregatePath(this, path);
+			aggregatePathCache.put(path, aggregatePath);
+		}
+
+		return aggregatePath;
+	}
+
+	public AggregatePath getAggregatePath(RelationalPersistentEntity<?> type) {
+
+		AggregatePath aggregatePath = aggregatePathCache.get(type);
+		if (aggregatePath == null) {
+
+			aggregatePath = new DefaultAggregatePath(this, type);
+			aggregatePathCache.put(type, aggregatePath);
+		}
+
+		return aggregatePath;
+	}
 }

spring-data-relational/src/main/java/org/springframework/data/relational/core/mapping/RelationalPersistentProperty.java

@@ -20,7 +20,7 @@ import org.springframework.data.relational.core.sql.SqlIdentifier;
 import org.springframework.lang.Nullable;
 
 /**
- * A {@link PersistentProperty} with methods for additional JDBC/RDBMS related meta data.
+ * A {@link PersistentProperty} with methods for additional RDBMS related metadata based on columns.
  *
  * @author Jens Schauder
  * @author Oliver Gierke
@@ -38,8 +38,18 @@ public interface RelationalPersistentProperty extends PersistentProperty<Relatio
 	@Override
 	RelationalPersistentEntity<?> getOwner();
 
+	/**
+	 * @deprecated use {@link #getReverseColumnName(RelationalPersistentEntity)} instead
+	 */
+	@Deprecated(since = "3.2", forRemoval = true)
 	SqlIdentifier getReverseColumnName(PersistentPropertyPathExtension path);
 
+	/**
+	 * @param owner the owning entity.
+	 * @return the column name to represent the owning side.
+	 */
+	SqlIdentifier getReverseColumnName(RelationalPersistentEntity<?> owner);
+
 	@Nullable
 	SqlIdentifier getKeyColumn();
 
@@ -78,7 +88,7 @@ public interface RelationalPersistentProperty extends PersistentProperty<Relatio
 
 	/**
 	 * Returns whether this property is only to be used during inserts and read.
-	 * 
+	 *
 	 * @since 3.0
 	 */
 	boolean isInsertOnly();

spring-data-relational/src/test/java/org/springframework/data/relational/core/mapping/BasicRelationalPersistentPropertyUnitTests.java

@@ -15,12 +15,10 @@
  */
 package org.springframework.data.relational.core.mapping;
 
+import static org.assertj.core.api.Assertions.*;
+import static org.springframework.data.relational.core.sql.SqlIdentifier.*;
+
 import junit.framework.AssertionFailedError;
-import org.assertj.core.api.SoftAssertions;
-import org.junit.jupiter.api.Test;
-import org.springframework.data.annotation.Id;
-import org.springframework.data.mapping.PersistentPropertyPath;
-import org.springframework.data.relational.core.mapping.Embedded.OnEmpty;
 
 import java.time.LocalDateTime;
 import java.time.ZonedDateTime;
@@ -28,8 +26,11 @@ import java.util.Date;
 import java.util.List;
 import java.util.function.BiConsumer;
 
-import static org.assertj.core.api.Assertions.*;
-import static org.springframework.data.relational.core.sql.SqlIdentifier.*;
+import org.assertj.core.api.SoftAssertions;
+import org.junit.jupiter.api.Test;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.mapping.PersistentPropertyPath;
+import org.springframework.data.relational.core.mapping.Embedded.OnEmpty;
 
 /**
  * Unit tests for the {@link BasicRelationalPersistentProperty}.
@@ -62,7 +63,7 @@ public class BasicRelationalPersistentPropertyUnitTests {
 				.findPersistentPropertyPaths(DummyEntity.class, p -> p.getName().equals("someList")).getFirst()
 				.orElseThrow(() -> new AssertionFailedError("Couldn't find path for 'someList'"));
 
-		assertThat(listProperty.getReverseColumnName(new PersistentPropertyPathExtension(context, path)))
+		assertThat(listProperty.getReverseColumnName(path.getLeafProperty().getOwner()))
 				.isEqualTo(quoted("dummy_column_name"));
 		assertThat(listProperty.getKeyColumn()).isEqualTo(quoted("dummy_key_column_name"));
 	}
@@ -87,7 +88,6 @@ public class BasicRelationalPersistentPropertyUnitTests {
 		RelationalPersistentProperty property = entity.getRequiredPersistentProperty("someList");
 
 		assertThat(property.getKeyColumn()).isEqualTo(quoted("key_col"));
-		assertThat(property.getReverseColumnName(null)).isEqualTo(quoted("id_col"));
 	}
 
 	@Test // DATAJDBC-111

spring-data-relational/src/test/java/org/springframework/data/relational/core/mapping/DefaultAggregatePathUnitTests.java

@@ -0,0 +1,522 @@
+/*
+ * Copyright 2023 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.relational.core.mapping;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.assertj.core.api.SoftAssertions.*;
+import static org.springframework.data.relational.core.sql.SqlIdentifier.*;
+
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.annotation.ReadOnlyProperty;
+import org.springframework.data.mapping.PersistentPropertyPath;
+import org.springframework.data.relational.core.sql.SqlIdentifier;
+
+/**
+ * Tests for {@link AggregatePath}.
+ *
+ * @author Jens Schauder
+ * @author Mark Paluch
+ */
+class DefaultAggregatePathUnitTests {
+	RelationalMappingContext context = new RelationalMappingContext();
+
+	private RelationalPersistentEntity<?> entity = context.getRequiredPersistentEntity(DummyEntity.class);
+
+	@Test // GH-1525
+	void isNotRootForNonRootPath() {
+
+		AggregatePath path = context.getAggregatePath(context.getPersistentPropertyPath("entityId", DummyEntity.class));
+
+		assertThat(path.isRoot()).isFalse();
+	}
+
+	@Test // GH-1525
+	void isRootForRootPath() {
+
+		AggregatePath path = context.getAggregatePath(entity);
+
+		assertThat(path.isRoot()).isTrue();
+	}
+
+	@Test // GH-1525
+	void getParentPath() {
+
+		assertSoftly(softly -> {
+
+			softly.assertThat(path("second.third2.value").getParentPath()).isEqualTo(path("second.third2"));
+			softly.assertThat(path("second.third2").getParentPath()).isEqualTo(path("second"));
+			softly.assertThat(path("second").getParentPath()).isEqualTo(path());
+
+			softly.assertThatThrownBy(() -> path().getParentPath()).isInstanceOf(IllegalStateException.class);
+		});
+	}
+
+	@Test // GH-1525
+	void getRequiredLeafEntity() {
+
+		assertSoftly(softly -> {
+
+			softly.assertThat(path().getRequiredLeafEntity()).isEqualTo(entity);
+			softly.assertThat(path("second").getRequiredLeafEntity())
+					.isEqualTo(context.getRequiredPersistentEntity(Second.class));
+			softly.assertThat(path("second.third").getRequiredLeafEntity())
+					.isEqualTo(context.getRequiredPersistentEntity(Third.class));
+			softly.assertThat(path("secondList").getRequiredLeafEntity())
+					.isEqualTo(context.getRequiredPersistentEntity(Second.class));
+
+			softly.assertThatThrownBy(() -> path("secondList.third.value").getRequiredLeafEntity())
+					.isInstanceOf(IllegalStateException.class);
+
+		});
+	}
+
+	@Test // GH-1525
+	void idDefiningPath() {
+
+		assertSoftly(softly -> {
+
+			softly.assertThat(path("second.third2.value").getIdDefiningParentPath()).isEqualTo(path());
+			softly.assertThat(path("second.third.value").getIdDefiningParentPath()).isEqualTo(path());
+			softly.assertThat(path("secondList.third2.value").getIdDefiningParentPath()).isEqualTo(path());
+			softly.assertThat(path("secondList.third.value").getIdDefiningParentPath()).isEqualTo(path());
+			softly.assertThat(path("second2.third2.value").getIdDefiningParentPath()).isEqualTo(path());
+			softly.assertThat(path("second2.third.value").getIdDefiningParentPath()).isEqualTo(path());
+			softly.assertThat(path("withId.second.third2.value").getIdDefiningParentPath()).isEqualTo(path("withId"));
+			softly.assertThat(path("withId.second.third.value").getIdDefiningParentPath()).isEqualTo(path("withId"));
+		});
+	}
+
+	@Test // GH-1525
+	void getRequiredIdProperty() {
+
+		assertSoftly(softly -> {
+
+			softly.assertThat(path().getRequiredIdProperty().getName()).isEqualTo("entityId");
+			softly.assertThat(path("withId").getRequiredIdProperty().getName()).isEqualTo("withIdId");
+			softly.assertThatThrownBy(() -> path("second").getRequiredIdProperty()).isInstanceOf(IllegalStateException.class);
+		});
+	}
+
+	@Test // GH-1525
+	void reverseColumnName() {
+
+		assertSoftly(softly -> {
+
+			softly.assertThat(path("second.third2").getTableInfo().reverseColumnInfo().name())
+					.isEqualTo(quoted("DUMMY_ENTITY"));
+			softly.assertThat(path("second.third").getTableInfo().reverseColumnInfo().name())
+					.isEqualTo(quoted("DUMMY_ENTITY"));
+			softly.assertThat(path("secondList.third2").getTableInfo().reverseColumnInfo().name())
+					.isEqualTo(quoted("DUMMY_ENTITY"));
+			softly.assertThat(path("secondList.third").getTableInfo().reverseColumnInfo().name())
+					.isEqualTo(quoted("DUMMY_ENTITY"));
+			softly.assertThat(path("second2.third").getTableInfo().reverseColumnInfo().name())
+					.isEqualTo(quoted("DUMMY_ENTITY"));
+			softly.assertThat(path("withId.second.third2.value").getTableInfo().reverseColumnInfo().name())
+					.isEqualTo(quoted("WITH_ID"));
+			softly.assertThat(path("withId.second.third").getTableInfo().reverseColumnInfo().name())
+					.isEqualTo(quoted("WITH_ID"));
+			softly.assertThat(path("withId.second2.third").getTableInfo().reverseColumnInfo().name())
+					.isEqualTo(quoted("WITH_ID"));
+		});
+	}
+
+	@Test // GH-1525
+	void getQualifierColumn() {
+
+		assertSoftly(softly -> {
+
+			softly.assertThat(path().getTableInfo().qualifierColumnInfo()).isEqualTo(null);
+			softly.assertThat(path("second.third").getTableInfo().qualifierColumnInfo()).isEqualTo(null);
+			softly.assertThat(path("secondList.third2").getTableInfo().qualifierColumnInfo()).isEqualTo(null);
+			softly.assertThat(path("secondList").getTableInfo().qualifierColumnInfo().name())
+					.isEqualTo(SqlIdentifier.quoted("DUMMY_ENTITY_KEY"));
+
+		});
+	}
+
+	@Test // GH-1525
+	void getQualifierColumnType() {
+
+		assertSoftly(softly -> {
+
+			softly.assertThat(path().getTableInfo().qualifierColumnType()).isEqualTo(null);
+			softly.assertThat(path("second.third").getTableInfo().qualifierColumnType()).isEqualTo(null);
+			softly.assertThat(path("secondList.third2").getTableInfo().qualifierColumnType()).isEqualTo(null);
+			softly.assertThat(path("secondList").getTableInfo().qualifierColumnType()).isEqualTo(Integer.class);
+
+		});
+	}
+
+	@Test // GH-1525
+	void extendBy() {
+
+		assertSoftly(softly -> {
+
+			softly.assertThat(path().append(entity.getRequiredPersistentProperty("withId"))).isEqualTo(path("withId"));
+			softly.assertThat(path("withId").append(path("withId").getRequiredIdProperty()))
+					.isEqualTo(path("withId.withIdId"));
+		});
+	}
+
+	@Test // GH-1525
+	void isWritable() {
+
+		assertSoftly(softly -> {
+			softly.assertThat(context.getAggregatePath(createSimplePath("withId")).isWritable())
+					.describedAs("simple path is writable").isTrue();
+			softly.assertThat(context.getAggregatePath(createSimplePath("secondList.third2")).isWritable())
+					.describedAs("long path is writable").isTrue();
+			softly.assertThat(context.getAggregatePath(createSimplePath("second")).isWritable())
+					.describedAs("simple read only path is not writable").isFalse();
+			softly.assertThat(context.getAggregatePath(createSimplePath("second.third")).isWritable())
+					.describedAs("long path containing read only element is not writable").isFalse();
+		});
+	}
+
+	@Test // GH-1525
+	void isEmbedded() {
+
+		assertSoftly(softly -> {
+			softly.assertThat(path().isEmbedded()).isFalse();
+			softly.assertThat(path("withId").isEmbedded()).isFalse();
+			softly.assertThat(path("second2.third").isEmbedded()).isFalse();
+			softly.assertThat(path("second2").isEmbedded()).isTrue();
+
+		});
+	}
+
+	@Test // GH-1525
+	void isEntity() {
+
+		assertSoftly(softly -> {
+
+			softly.assertThat(path().isEntity()).isTrue();
+			softly.assertThat(path("second").isEntity()).isTrue();
+			softly.assertThat(path("second.third2").isEntity()).isTrue();
+			softly.assertThat(path("secondList.third2").isEntity()).isTrue();
+			softly.assertThat(path("secondList").isEntity()).isTrue();
+			softly.assertThat(path("second.third2.value").isEntity()).isFalse();
+			softly.assertThat(path("secondList.third2.value").isEntity()).isFalse();
+		});
+	}
+
+	@Test // GH-1525
+	void isMultiValued() {
+
+		assertSoftly(softly -> {
+
+			softly.assertThat(path().isMultiValued()).isFalse();
+			softly.assertThat(path("second").isMultiValued()).isFalse();
+			softly.assertThat(path("second.third2").isMultiValued()).isFalse();
+			softly.assertThat(path("secondList.third2").isMultiValued()).isTrue(); // this seems wrong as third2 is an
+																																							// embedded path into Second, held by
+																																							// List<Second> (so the parent is
+																																							// multi-valued but not third2).
+			// TODO: This test fails because MultiValued considers parents.
+			// softly.assertThat(path("secondList.third.value").isMultiValued()).isFalse();
+			softly.assertThat(path("secondList").isMultiValued()).isTrue();
+		});
+	}
+
+	@Test // GH-1525
+	void isQualified() {
+
+		assertSoftly(softly -> {
+
+			softly.assertThat(path().isQualified()).isFalse();
+			softly.assertThat(path("second").isQualified()).isFalse();
+			softly.assertThat(path("second.third2").isQualified()).isFalse();
+			softly.assertThat(path("secondList.third2").isQualified()).isFalse();
+			softly.assertThat(path("secondList").isQualified()).isTrue();
+		});
+	}
+
+	@Test // GH-1525
+	void isMap() {
+
+		assertSoftly(softly -> {
+
+			softly.assertThat(path().isMap()).isFalse();
+			softly.assertThat(path("second").isMap()).isFalse();
+			softly.assertThat(path("second.third2").isMap()).isFalse();
+			softly.assertThat(path("secondList.third2").isMap()).isFalse();
+			softly.assertThat(path("secondList").isMap()).isFalse();
+			softly.assertThat(path("secondMap.third2").isMap()).isFalse();
+			softly.assertThat(path("secondMap").isMap()).isTrue();
+		});
+	}
+
+	@Test // GH-1525
+	void isCollectionLike() {
+
+		assertSoftly(softly -> {
+
+			softly.assertThat(path().isCollectionLike()).isFalse();
+			softly.assertThat(path("second").isCollectionLike()).isFalse();
+			softly.assertThat(path("second.third2").isCollectionLike()).isFalse();
+			softly.assertThat(path("secondList.third2").isCollectionLike()).isFalse();
+			softly.assertThat(path("secondMap.third2").isCollectionLike()).isFalse();
+			softly.assertThat(path("secondMap").isCollectionLike()).isFalse();
+			softly.assertThat(path("secondList").isCollectionLike()).isTrue();
+		});
+	}
+
+	@Test // GH-1525
+	void isOrdered() {
+
+		assertSoftly(softly -> {
+
+			softly.assertThat(path().isOrdered()).isFalse();
+			softly.assertThat(path("second").isOrdered()).isFalse();
+			softly.assertThat(path("second.third2").isOrdered()).isFalse();
+			softly.assertThat(path("secondList.third2").isOrdered()).isFalse();
+			softly.assertThat(path("secondMap.third2").isOrdered()).isFalse();
+			softly.assertThat(path("secondMap").isOrdered()).isFalse();
+			softly.assertThat(path("secondList").isOrdered()).isTrue();
+		});
+	}
+
+	@Test // GH-1525
+	void getTableAlias() {
+
+		assertSoftly(softly -> {
+
+			softly.assertThat(path().getTableInfo().tableAlias()).isEqualTo(null);
+			softly.assertThat(path("second").getTableInfo().tableAlias()).isEqualTo(quoted("second"));
+			softly.assertThat(path("second.third2").getTableInfo().tableAlias()).isEqualTo(quoted("second"));
+			softly.assertThat(path("second.third2.value").getTableInfo().tableAlias()).isEqualTo(quoted("second"));
+			softly.assertThat(path("second.third").getTableInfo().tableAlias()).isEqualTo(quoted("second_third")); // missing
+																																																							// _
+			softly.assertThat(path("second.third.value").getTableInfo().tableAlias()).isEqualTo(quoted("second_third")); // missing
+																																																										// _
+			softly.assertThat(path("secondList.third2").getTableInfo().tableAlias()).isEqualTo(quoted("secondList"));
+			softly.assertThat(path("secondList.third2.value").getTableInfo().tableAlias()).isEqualTo(quoted("secondList"));
+			softly.assertThat(path("secondList.third").getTableInfo().tableAlias()).isEqualTo(quoted("secondList_third")); // missing
+																																																											// _
+			softly.assertThat(path("secondList.third.value").getTableInfo().tableAlias())
+					.isEqualTo(quoted("secondList_third")); // missing _
+			softly.assertThat(path("secondList").getTableInfo().tableAlias()).isEqualTo(quoted("secondList"));
+			softly.assertThat(path("second2.third").getTableInfo().tableAlias()).isEqualTo(quoted("secthird"));
+			softly.assertThat(path("second3.third").getTableInfo().tableAlias()).isEqualTo(quoted("third"));
+		});
+	}
+
+	@Test // GH-1525
+	void getTableName() {
+
+		assertSoftly(softly -> {
+
+			softly.assertThat(path().getTableInfo().qualifiedTableName()).isEqualTo(quoted("DUMMY_ENTITY"));
+			softly.assertThat(path("second").getTableInfo().qualifiedTableName()).isEqualTo(quoted("SECOND"));
+			softly.assertThat(path("second.third2").getTableInfo().qualifiedTableName()).isEqualTo(quoted("SECOND"));
+			softly.assertThat(path("second.third2.value").getTableInfo().qualifiedTableName()).isEqualTo(quoted("SECOND"));
+			softly.assertThat(path("secondList.third2").getTableInfo().qualifiedTableName()).isEqualTo(quoted("SECOND"));
+			softly.assertThat(path("secondList.third2.value").getTableInfo().qualifiedTableName())
+					.isEqualTo(quoted("SECOND"));
+			softly.assertThat(path("secondList").getTableInfo().qualifiedTableName()).isEqualTo(quoted("SECOND"));
+		});
+	}
+
+	@Test // GH-1525
+	void getColumnName() {
+
+		assertSoftly(softly -> {
+
+			softly.assertThat(path("second.third2.value").getColumnInfo().name()).isEqualTo(quoted("THRDVALUE"));
+			softly.assertThat(path("second.third.value").getColumnInfo().name()).isEqualTo(quoted("VALUE"));
+			softly.assertThat(path("secondList.third2.value").getColumnInfo().name()).isEqualTo(quoted("THRDVALUE"));
+			softly.assertThat(path("secondList.third.value").getColumnInfo().name()).isEqualTo(quoted("VALUE"));
+			softly.assertThat(path("second2.third2.value").getColumnInfo().name()).isEqualTo(quoted("SECTHRDVALUE"));
+			softly.assertThat(path("second2.third.value").getColumnInfo().name()).isEqualTo(quoted("VALUE"));
+		});
+	}
+
+	@Test // GH-1525
+	void getColumnAlias() {
+
+		assertSoftly(softly -> {
+
+			softly.assertThat(path("second.third2.value").getColumnInfo().alias()).isEqualTo(quoted("SECOND_THRDVALUE"));
+			softly.assertThat(path("second.third.value").getColumnInfo().alias()).isEqualTo(quoted("SECOND_THIRD_VALUE"));
+			softly.assertThat(path("secondList.third2.value").getColumnInfo().alias())
+					.isEqualTo(quoted("SECONDLIST_THRDVALUE"));
+			softly.assertThat(path("secondList.third.value").getColumnInfo().alias())
+					.isEqualTo(quoted("SECONDLIST_THIRD_VALUE"));
+			softly.assertThat(path("second2.third2.value").getColumnInfo().alias()).isEqualTo(quoted("SECTHRDVALUE"));
+			softly.assertThat(path("second2.third.value").getColumnInfo().alias()).isEqualTo(quoted("SECTHIRD_VALUE"));
+		});
+	}
+
+	@Test // GH-1525
+	void getReverseColumnAlias() {
+
+		assertSoftly(softly -> {
+
+			softly.assertThat(path("second.third2.value").getTableInfo().reverseColumnInfo().alias())
+					.isEqualTo(quoted("SECOND_DUMMY_ENTITY"));
+			softly.assertThat(path("second.third.value").getTableInfo().reverseColumnInfo().alias())
+					.isEqualTo(quoted("SECOND_THIRD_DUMMY_ENTITY"));
+			softly.assertThat(path("secondList.third2.value").getTableInfo().reverseColumnInfo().alias())
+					.isEqualTo(quoted("SECONDLIST_DUMMY_ENTITY"));
+			softly.assertThat(path("secondList.third.value").getTableInfo().reverseColumnInfo().alias())
+					.isEqualTo(quoted("SECONDLIST_THIRD_DUMMY_ENTITY"));
+			softly.assertThat(path("second2.third.value").getTableInfo().reverseColumnInfo().alias())
+					.isEqualTo(quoted("SECTHIRD_DUMMY_ENTITY"));
+		});
+	}
+
+	@Test // GH-1525
+	void getRequiredLeafProperty() {
+
+		assertSoftly(softly -> {
+
+			softly.assertThat(path("second.third2.value").getRequiredLeafProperty())
+					.isEqualTo(context.getRequiredPersistentEntity(Third.class).getPersistentProperty("value"));
+			softly.assertThat(path("second.third").getRequiredLeafProperty())
+					.isEqualTo(context.getRequiredPersistentEntity(Second.class).getPersistentProperty("third"));
+			softly.assertThat(path("secondList").getRequiredLeafProperty())
+					.isEqualTo(entity.getPersistentProperty("secondList"));
+			softly.assertThatThrownBy(() -> path().getRequiredLeafProperty()).isInstanceOf(IllegalStateException.class);
+		});
+	}
+
+	@Test // GH-1525
+	void getBaseProperty() {
+
+		assertSoftly(softly -> {
+
+			softly.assertThat(path("second.third2.value").getRequiredBaseProperty())
+					.isEqualTo(entity.getPersistentProperty("second"));
+			softly.assertThat(path("second.third.value").getRequiredBaseProperty())
+					.isEqualTo(entity.getPersistentProperty("second"));
+			softly.assertThat(path("secondList.third2.value").getRequiredBaseProperty())
+					.isEqualTo(entity.getPersistentProperty("secondList"));
+			softly.assertThatThrownBy(() -> path().getRequiredBaseProperty()).isInstanceOf(IllegalStateException.class);
+		});
+	}
+
+	@Test // GH-1525
+	void getIdColumnName() {
+
+		assertSoftly(softly -> {
+
+			softly.assertThat(path().getTableInfo().idColumnName()).isEqualTo(quoted("ENTITY_ID"));
+			softly.assertThat(path("withId").getTableInfo().idColumnName()).isEqualTo(quoted("WITH_ID_ID"));
+
+			softly.assertThat(path("second").getTableInfo().idColumnName()).isNull();
+			softly.assertThat(path("second.third2").getTableInfo().idColumnName()).isNull();
+			softly.assertThat(path("withId.second").getTableInfo().idColumnName()).isNull();
+		});
+	}
+
+	@Test // GH-1525
+	void toDotPath() {
+
+		assertSoftly(softly -> {
+
+			softly.assertThat(path().toDotPath()).isEqualTo("");
+			softly.assertThat(path("second.third.value").toDotPath()).isEqualTo("second.third.value");
+		});
+	}
+
+	@Test // GH-1525
+	void getRequiredPersistentPropertyPath() {
+
+		assertSoftly(softly -> {
+
+			softly.assertThat(path("second.third.value").getRequiredPersistentPropertyPath())
+					.isEqualTo(createSimplePath("second.third.value"));
+			softly.assertThatThrownBy(() -> path().getRequiredPersistentPropertyPath())
+					.isInstanceOf(IllegalStateException.class);
+		});
+	}
+
+	@Test // GH-1525
+	void getEffectiveIdColumnName() {
+
+		assertSoftly(softly -> {
+
+			softly.assertThat(path().getTableInfo().effectiveIdColumnName()).isEqualTo(quoted("ENTITY_ID"));
+			softly.assertThat(path("second.third2").getTableInfo().effectiveIdColumnName()).isEqualTo(quoted("DUMMY_ENTITY"));
+			softly.assertThat(path("withId.second.third").getTableInfo().effectiveIdColumnName())
+					.isEqualTo(quoted("WITH_ID"));
+			softly.assertThat(path("withId.second.third2.value").getTableInfo().effectiveIdColumnName())
+					.isEqualTo(quoted("WITH_ID"));
+		});
+	}
+
+	@Test // GH-1525
+	void getLength() {
+
+		assertThat(path().getLength()).isEqualTo(1);
+		assertThat(path().stream().collect(Collectors.toList())).hasSize(1);
+
+		assertThat(path("second.third2").getLength()).isEqualTo(3);
+		assertThat(path("second.third2").stream().collect(Collectors.toList())).hasSize(3);
+
+		assertThat(path("withId.second.third").getLength()).isEqualTo(4);
+		assertThat(path("withId.second.third2.value").getLength()).isEqualTo(5);
+	}
+
+	private AggregatePath path() {
+		return context.getAggregatePath(entity);
+	}
+
+	private AggregatePath path(String path) {
+		return context.getAggregatePath(createSimplePath(path));
+	}
+
+	PersistentPropertyPath<RelationalPersistentProperty> createSimplePath(String path) {
+		return PersistentPropertyPathTestUtils.getPath(context, path, DummyEntity.class);
+	}
+
+	@SuppressWarnings("unused")
+	static class DummyEntity {
+		@Id Long entityId;
+		@ReadOnlyProperty Second second;
+		@Embedded(onEmpty = Embedded.OnEmpty.USE_NULL, prefix = "sec") Second second2;
+		@Embedded(onEmpty = Embedded.OnEmpty.USE_NULL) Second second3;
+		List<Second> secondList;
+		Map<String, Second> secondMap;
+		WithId withId;
+	}
+
+	@SuppressWarnings("unused")
+	static class Second {
+		Third third;
+		@Embedded(onEmpty = Embedded.OnEmpty.USE_NULL, prefix = "thrd") Third third2;
+	}
+
+	@SuppressWarnings("unused")
+	static class Third {
+		String value;
+	}
+
+	@SuppressWarnings("unused")
+	static class WithId {
+		@Id Long withIdId;
+		Second second;
+		@Embedded(onEmpty = Embedded.OnEmpty.USE_NULL, prefix = "sec") Second second2;
+	}
+
+}

spring-data-relational/src/test/java/org/springframework/data/relational/core/mapping/PersistentPropertyPathTestUtils.java

@@ -13,9 +13,10 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.springframework.data.jdbc.core.mapping;
+package org.springframework.data.relational.core.mapping;
 
 import org.springframework.data.mapping.PersistentPropertyPath;
+import org.springframework.data.mapping.PersistentPropertyPaths;
 import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
 
@@ -31,10 +32,14 @@ public final class PersistentPropertyPathTestUtils {
 	public static PersistentPropertyPath<RelationalPersistentProperty> getPath(RelationalMappingContext context,
 																			   String path, Class<?> baseType) {
 
-		return context.findPersistentPropertyPaths(baseType, p -> p.isEntity()) //
-				.filter(p -> p.toDotPath().equals(path)) //
-				.stream() //
-				.findFirst() //
-				.orElseThrow(() -> new IllegalArgumentException(String.format("No path for %s based on %s", path, baseType)));
+		PersistentPropertyPaths<?, RelationalPersistentProperty> persistentPropertyPaths = context
+				.findPersistentPropertyPaths(baseType, p -> true);
+
+		return persistentPropertyPaths
+				.filter(p -> p.toDotPath().equals(path))
+				.stream()
+				.findFirst()
+				.orElse(null);
+
 	}
 }

spring-data-relational/src/test/java/org/springframework/data/relational/core/mapping/RelationalMappingContextUnitTests.java

@@ -21,8 +21,10 @@ import java.util.Arrays;
 import java.util.HashSet;
 import java.util.UUID;
 
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.springframework.data.annotation.Id;
+import org.springframework.data.mapping.PersistentPropertyPath;
 import org.springframework.data.mapping.model.SimpleTypeHolder;
 
 /**
@@ -31,21 +33,60 @@ import org.springframework.data.mapping.model.SimpleTypeHolder;
  * @author Toshiaki Maki
  */
 public class RelationalMappingContextUnitTests {
+	RelationalMappingContext context = new RelationalMappingContext();
+	SimpleTypeHolder holder = new SimpleTypeHolder(new HashSet<>(Arrays.asList(UUID.class)), true);
+
+	@BeforeEach
+	void setup() {
+		context.setSimpleTypeHolder(holder);
+	}
 
 	@Test // DATAJDBC-229
 	public void uuidPropertyIsNotEntity() {
 
-		SimpleTypeHolder holder = new SimpleTypeHolder(new HashSet<>(Arrays.asList(UUID.class)), true);
-
-		RelationalMappingContext mappingContext = new RelationalMappingContext();
-		mappingContext.setSimpleTypeHolder(holder);
-
-		RelationalPersistentEntity<?> entity = mappingContext.getPersistentEntity(EntityWithUuid.class);
+		RelationalPersistentEntity<?> entity = context.getPersistentEntity(EntityWithUuid.class);
 		RelationalPersistentProperty uuidProperty = entity.getRequiredPersistentProperty("uuid");
 
 		assertThat(uuidProperty.isEntity()).isFalse();
 	}
 
+	@Test // GH-1525
+	public void canObtainAggregatePath() {
+
+		PersistentPropertyPath<RelationalPersistentProperty> path = context.getPersistentPropertyPath("uuid",
+				EntityWithUuid.class);
+		AggregatePath aggregatePath = context.getAggregatePath(path);
+
+		assertThat(aggregatePath).isNotNull();
+	}
+
+	@Test // GH-1525
+	public void innerAggregatePathsGetCached() {
+
+		context = new RelationalMappingContext();
+		context.setSimpleTypeHolder(holder);
+
+		PersistentPropertyPath<RelationalPersistentProperty> path = context.getPersistentPropertyPath("uuid",
+				EntityWithUuid.class);
+
+		AggregatePath one = context.getAggregatePath(path);
+		AggregatePath two = context.getAggregatePath(path);
+
+		assertThat(one).isSameAs(two);
+	}
+
+	@Test // GH-1525
+	public void rootAggregatePathsGetCached() {
+
+		context = new RelationalMappingContext();
+		context.setSimpleTypeHolder(holder);
+
+		AggregatePath one = context.getAggregatePath(context.getRequiredPersistentEntity(EntityWithUuid.class));
+		AggregatePath two = context.getAggregatePath(context.getRequiredPersistentEntity(EntityWithUuid.class));
+
+		assertThat(one).isSameAs(two);
+	}
+
 	static class EntityWithUuid {
 		@Id UUID uuid;
 	}