Applied changes requested in review.

The major ones are:

* directly construct joins
* remove multiple places of duplication
* naming
* documentation

See #574
Original pull request #1957

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

@@ -17,6 +17,7 @@ package org.springframework.data.jdbc.core;
 
 import java.util.*;
 import java.util.function.BiConsumer;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 
 import org.springframework.dao.IncorrectUpdateSemanticsDataAccessException;
@@ -176,7 +177,8 @@ class JdbcAggregateChangeExecutionContext {
 		Object id = getParentId(action);
 
 		JdbcIdentifierBuilder identifier = JdbcIdentifierBuilder //
-				.forBackReferences(converter, context.getAggregatePath(action.getPropertyPath()), id);
+				.forBackReferences(converter, context.getAggregatePath(action.getPropertyPath()),
+						getValueProvider(id, context.getAggregatePath(action.getPropertyPath()), converter));
 
 		for (Map.Entry<PersistentPropertyPath<RelationalPersistentProperty>, Object> qualifier : action.getQualifiers()
 				.entrySet()) {
@@ -186,6 +188,22 @@ class JdbcAggregateChangeExecutionContext {
 		return identifier.build();
 	}
 
+	static Function<AggregatePath, Object> getValueProvider(Object idValue, AggregatePath path, JdbcConverter converter) {
+
+		RelationalPersistentEntity<?> entity = converter.getMappingContext()
+				.getPersistentEntity(path.getIdDefiningParentPath().getRequiredIdProperty().getType());
+
+		Function<AggregatePath, Object> valueProvider = ap -> {
+			if (entity == null) {
+				return idValue;
+			} else {
+				PersistentPropertyPathAccessor<Object> propertyPathAccessor = entity.getPropertyPathAccessor(idValue);
+				return propertyPathAccessor.getProperty(ap.getRequiredPersistentPropertyPath());
+			}
+		};
+		return valueProvider;
+	}
+
 	private Object getParentId(DbAction.WithDependingOn<?> action) {
 
 		DbAction.WithEntity<?> idOwningAction = getIdOwningAction(action,
@@ -267,12 +285,10 @@ class JdbcAggregateChangeExecutionContext {
 
 				if (newEntity != action.getEntity()) {
 
-					cascadingValues.stage(insert.getDependingOn(), insert.getPropertyPath(),
+					cascadingValues.stage(insert.getDependingOn(), insert.getPropertyPath(), qualifierValue, newEntity);
-							qualifierValue, newEntity);
 				} else if (insert.getPropertyPath().getLeafProperty().isCollectionLike()) {
 
-					cascadingValues.gather(insert.getDependingOn(), insert.getPropertyPath(),
+					cascadingValues.gather(insert.getDependingOn(), insert.getPropertyPath(), qualifierValue, newEntity);
-							qualifierValue, newEntity);
 				}
 			}
 		}
@@ -359,8 +375,8 @@ class JdbcAggregateChangeExecutionContext {
 	 */
 	private static class StagedValues {
 
-		static final List<MultiValueAggregator<?>> aggregators = Arrays.asList(SetAggregator.INSTANCE, MapAggregator.INSTANCE,
+		static final List<MultiValueAggregator<?>> aggregators = Arrays.asList(SetAggregator.INSTANCE,
-				ListAggregator.INSTANCE, SingleElementAggregator.INSTANCE);
+				MapAggregator.INSTANCE, ListAggregator.INSTANCE, SingleElementAggregator.INSTANCE);
 
 		Map<DbAction, Map<PersistentPropertyPath, StagedValue>> values = new HashMap<>();
 

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

@@ -106,7 +106,7 @@ public final class Identifier {
 	 * @param identifier the identifier to append.
 	 * @return the {@link Identifier} containing all existing keys and the key part for {@code name}, {@code value}, and a
 	 *         {@link Class target type}.
-	 * @since 3.5
+	 * @since 4.0
 	 */
 	public Identifier withPart(Identifier identifier) {
 
@@ -207,7 +207,6 @@ public final class Identifier {
 		return null;
 	}
 
-
 	/**
 	 * A single value of an Identifier consisting of the column name, the value and the target type which is to be used to
 	 * store the element in the database.

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

@@ -17,10 +17,7 @@ package org.springframework.data.jdbc.core.convert;
 
 import java.util.function.Function;
 
-import org.springframework.data.mapping.PersistentPropertyPathAccessor;
 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.util.Assert;
 
@@ -45,31 +42,42 @@ public class JdbcIdentifierBuilder {
 	/**
 	 * Creates ParentKeys with backreference for the given path and value of the parents id.
 	 */
-	public static JdbcIdentifierBuilder forBackReferences(JdbcConverter converter, AggregatePath path, Object value) {
+	public static JdbcIdentifierBuilder forBackReferences(JdbcConverter converter, AggregatePath path,
+			Function<AggregatePath, Object> valueProvider) {
 
-		RelationalPersistentProperty idProperty = path.getIdDefiningParentPath().getRequiredIdProperty();
+		return new JdbcIdentifierBuilder(forBackReference(converter, path, Identifier.empty(), valueProvider));
-		AggregatePath.ColumnInfos infos = path.getTableInfo().reverseColumnInfos();
+	}
 
-		// create property accessor
+	/**
-		RelationalMappingContext mappingContext = converter.getMappingContext();
+	 * @param converter used for determining the column types to be used for different properties. Must not be
-		RelationalPersistentEntity<?> persistentEntity = mappingContext.getPersistentEntity(idProperty.getType());
+	 *          {@literal null}.
+	 * @param path the path for which needs to back reference an id. Must not be {@literal null}.
+	 * @param defaultIdentifier Identifier to be used as a default when no backreference can be constructed. Must not be
+	 *          {@literal null}.
+	 * @param valueProvider provides values for the {@link Identifier} based on an {@link AggregatePath}. Must not be
+	 *          {@literal null}.
+	 * @return Guaranteed not to be {@literal null}.
+	 */
+	public static Identifier forBackReference(JdbcConverter converter, AggregatePath path, Identifier defaultIdentifier,
+			Function<AggregatePath, Object> valueProvider) {
 
-		Function<AggregatePath, Object> valueProvider;
+		Identifier identifierToUse = defaultIdentifier;
-		if (persistentEntity == null) {
-			valueProvider = ap -> value;
-		} else {
-			PersistentPropertyPathAccessor<Object> propertyPathAccessor = persistentEntity.getPropertyPathAccessor(value);
-			valueProvider = ap -> propertyPathAccessor.getProperty(ap.getRequiredPersistentPropertyPath());
-		}
 
-		Identifier identifierHolder = infos.reduce(Identifier.empty(), (ap, ci) -> {
+		AggregatePath idDefiningParentPath = path.getIdDefiningParentPath();
 
-			RelationalPersistentProperty property = ap.getRequiredLeafProperty();
+		// note that the idDefiningParentPath might not itself have an id property, but have a combination of back
-			return Identifier.of(ci.name(), valueProvider.apply(ap),
+		// references and possibly keys, that form an id
-					converter.getColumnType(property));
+		if (idDefiningParentPath.hasIdProperty()) {
-		}, Identifier::withPart);
+
+			AggregatePath.ColumnInfos infos = path.getTableInfo().backReferenceColumnInfos();
+			identifierToUse = infos.reduce(Identifier.empty(), (ap, ci) -> {
+
+				RelationalPersistentProperty property = ap.getRequiredLeafProperty();
+				return Identifier.of(ci.name(), valueProvider.apply(ap), converter.getColumnType(property));
+			}, Identifier::withPart);
+		}
 
-		return new JdbcIdentifierBuilder(identifierHolder);
+		return identifierToUse;
 	}
 
 	/**
@@ -85,8 +93,7 @@ public class JdbcIdentifierBuilder {
 		Assert.notNull(value, "Value must not be null");
 
 		AggregatePath.TableInfo tableInfo = path.getTableInfo();
-		identifier = identifier.withPart(tableInfo.qualifierColumnInfo().name(), value,
+		identifier = identifier.withPart(tableInfo.qualifierColumnInfo().name(), value, tableInfo.qualifierColumnType());
-				tableInfo.qualifierColumnType());
 
 		return this;
 	}

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

@@ -361,7 +361,8 @@ public class MappingJdbcConverter extends MappingRelationalConverter implements
 
 				if (property.isCollectionLike() || property.isMap()) {
 
-					Identifier identifier = constructIdentifier(aggregatePath);
+					Identifier identifier = JdbcIdentifierBuilder.forBackReference(MappingJdbcConverter.this, aggregatePath,
+							this.identifier, getWrappedValueProvider(delegate::getValue, aggregatePath));
 
 					Iterable<Object> allByPath = relationResolver.findAllByPath(identifier,
 							aggregatePath.getRequiredPersistentPropertyPath());
@@ -388,29 +389,6 @@ public class MappingJdbcConverter extends MappingRelationalConverter implements
 			return (T) delegate.getValue(aggregatePath);
 		}
 
-		private Identifier constructIdentifier(AggregatePath aggregatePath) {
-
-			Identifier identifierToUse = this.identifier;
-			AggregatePath idDefiningParentPath = aggregatePath.getIdDefiningParentPath();
-
-			// note that the idDefiningParentPath might not itself have an id property, but have a combination of back
-			// references and possibly keys, that form an id
-			if (idDefiningParentPath.hasIdProperty()) {
-
-				RelationalPersistentProperty idProperty = idDefiningParentPath.getRequiredIdProperty();
-				AggregatePath idPath = idProperty.isEntity() ? idDefiningParentPath.append(idProperty) : idDefiningParentPath;
-				Identifier[] buildingIdentifier = new Identifier[] { Identifier.empty() };
-				aggregatePath.getTableInfo().reverseColumnInfos().forEach((ap, ci) -> {
-
-					Object value = delegate.getValue(idPath.append(ap));
-					buildingIdentifier[0] = buildingIdentifier[0].withPart(ci.name(), value,
-							ap.getRequiredLeafProperty().getActualType());
-				});
-				identifierToUse = buildingIdentifier[0];
-			}
-			return identifierToUse;
-		}
-
 		@Override
 		public boolean hasValue(RelationalPersistentProperty property) {
 
@@ -431,7 +409,7 @@ public class MappingJdbcConverter extends MappingRelationalConverter implements
 					return delegate.hasValue(toUse);
 				}
 
-				return delegate.hasValue(aggregatePath.getTableInfo().reverseColumnInfos().any().alias());
+				return delegate.hasValue(aggregatePath.getTableInfo().backReferenceColumnInfos().any().alias());
 			}
 
 			return delegate.hasValue(aggregatePath);
@@ -457,7 +435,7 @@ public class MappingJdbcConverter extends MappingRelationalConverter implements
 					return delegate.hasValue(toUse);
 				}
 
-				return delegate.hasValue(aggregatePath.getTableInfo().reverseColumnInfos().any().alias());
+				return delegate.hasValue(aggregatePath.getTableInfo().backReferenceColumnInfos().any().alias());
 			}
 
 			return delegate.hasNonEmptyValue(aggregatePath);
@@ -472,6 +450,22 @@ public class MappingJdbcConverter extends MappingRelationalConverter implements
 		}
 	}
 
+	private static Function<AggregatePath, Object> getWrappedValueProvider(Function<AggregatePath, Object> valueProvider,
+			AggregatePath aggregatePath) {
+
+		AggregatePath idDefiningParentPath = aggregatePath.getIdDefiningParentPath();
+
+		if (!idDefiningParentPath.hasIdProperty()) {
+			return ap -> {
+				throw new IllegalStateException("This should never happen");
+			};
+		}
+
+		RelationalPersistentProperty idProperty = idDefiningParentPath.getRequiredIdProperty();
+		AggregatePath idPath = idProperty.isEntity() ? idDefiningParentPath.append(idProperty) : idDefiningParentPath;
+		return ap -> valueProvider.apply(idPath.append(ap));
+	}
+
 	/**
 	 * Marker object to indicate that the property value provider should resolve relations.
 	 *

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

@@ -66,10 +66,11 @@ class SqlContext {
 	 * 
 	 * @param path must not be null.
 	 * @return a {@literal Column} that is part of the effective primary key for the given path.
+	 * @since 4.0
 	 */
 	Column getAnyReverseColumn(AggregatePath path) {
 
-		AggregatePath.ColumnInfo columnInfo = path.getTableInfo().reverseColumnInfos().any();
+		AggregatePath.ColumnInfo columnInfo = path.getTableInfo().backReferenceColumnInfos().any();
 		return getTable(path).column(columnInfo.name()).as(columnInfo.alias());
 	}
 }

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

@@ -17,6 +17,7 @@ package org.springframework.data.jdbc.core.convert;
 
 import java.util.*;
 import java.util.function.Function;
+import java.util.function.Predicate;
 import java.util.stream.Collectors;
 
 import org.springframework.data.domain.Pageable;
@@ -34,7 +35,6 @@ import org.springframework.data.relational.core.mapping.RelationalPersistentProp
 import org.springframework.data.relational.core.query.CriteriaDefinition;
 import org.springframework.data.relational.core.query.Query;
 import org.springframework.data.relational.core.sql.*;
-import org.springframework.data.relational.core.sql.render.RenderContext;
 import org.springframework.data.relational.core.sql.render.SqlRenderer;
 import org.springframework.data.util.Lazy;
 import org.springframework.data.util.Pair;
@@ -62,7 +62,7 @@ import org.springframework.util.Assert;
  * @author Viktor Ardelean
  * @author Kurt Niemi
  */
-class SqlGenerator {
+public class SqlGenerator {
 
 	static final SqlIdentifier VERSION_SQL_PARAMETER = SqlIdentifier.unquoted("___oldOptimisticLockingVersion");
 	static final SqlIdentifier IDS_SQL_PARAMETER = SqlIdentifier.unquoted("ids");
@@ -96,10 +96,6 @@ class SqlGenerator {
 	private final QueryMapper queryMapper;
 	private final Dialect dialect;
 
-	private final Function<Map<AggregatePath, Column>, Condition> inCondition;
-	private final Function<Map<AggregatePath, Column>, Condition> equalityCondition;
-	private final Function<Map<AggregatePath, Column>, Condition> notNullCondition;
-
 	/**
 	 * Create a new {@link SqlGenerator} given {@link RelationalMappingContext} and {@link RelationalPersistentEntity}.
 	 *
@@ -118,11 +114,19 @@ class SqlGenerator {
 		this.columns = new Columns(entity, mappingContext, converter);
 		this.queryMapper = new QueryMapper(converter);
 		this.dialect = dialect;
+	}
 
-		inCondition = inCondition();
+	/**
-		equalityCondition = equalityCondition();
+	 * Create a basic select structure with all the necessary joins
-		notNullCondition = isNotNullCondition();
+	 * 
-
+	 * @param table the table to base the select on
+	 * @param pathFilter a filter for excluding paths from the select. All paths for which the filter returns
+	 *          {@literal true} will be skipped when determining columns to select.
+	 * @return A select structure suitable for constructing more specialized selects by adding conditions.
+	 * @since 4.0
+	 */
+	public SelectBuilder.SelectWhere createSelectBuilder(Table table, Predicate<AggregatePath> pathFilter) {
+		return createSelectBuilder(table, pathFilter, Collections.emptyList());
 	}
 
 	/**
@@ -199,7 +203,7 @@ class SqlGenerator {
 			innerCondition = getSubselectCondition(parentPath, conditionFunction, selectFilterColumns);
 		}
 
-		List<Expression> idColumns = parentPathTableInfo.idColumnInfos().toList(ci -> subSelectTable.column(ci.name()));
+		List<Column> idColumns = parentPathTableInfo.idColumnInfos().toColumnList(subSelectTable);
 
 		Select select = Select.builder() //
 				.select(idColumns) //
@@ -210,7 +214,7 @@ class SqlGenerator {
 	}
 
 	private Expression toExpression(Map<AggregatePath, Column> columnsMap) {
-			return TupleExpression.maybeWrap(new ArrayList<>(columnsMap.values()));
+		return Expressions.of(new ArrayList<>(columnsMap.values()));
 	}
 
 	private BindMarker getBindMarker(SqlIdentifier columnName) {
@@ -456,7 +460,7 @@ class SqlGenerator {
 			return render(deleteAll.build());
 		}
 
-		return createDeleteByPathAndCriteria(mappingContext.getAggregatePath(path), notNullCondition);
+		return createDeleteByPathAndCriteria(mappingContext.getAggregatePath(path), this::isNotNullCondition);
 	}
 
 	/**
@@ -467,7 +471,7 @@ class SqlGenerator {
 	 * @return the statement as a {@link String}. Guaranteed to be not {@literal null}.
 	 */
 	String createDeleteByPath(PersistentPropertyPath<RelationalPersistentProperty> path) {
-		return createDeleteByPathAndCriteria(mappingContext.getAggregatePath(path), equalityCondition);
+		return createDeleteByPathAndCriteria(mappingContext.getAggregatePath(path), this::equalityCondition);
 	}
 
 	/**
@@ -478,63 +482,55 @@ class SqlGenerator {
 	 * @return the statement as a {@link String}. Guaranteed to be not {@literal null}.
 	 */
 	String createDeleteInByPath(PersistentPropertyPath<RelationalPersistentProperty> path) {
-		return createDeleteByPathAndCriteria(mappingContext.getAggregatePath(path), inCondition);
+		return createDeleteByPathAndCriteria(mappingContext.getAggregatePath(path), this::inCondition);
 	}
 
 	/**
-	 * Constructs a function for constructing a where condition. The where condition will be of the form
+	 * Constructs a where condition. The where condition will be of the form {@literal <columns> IN :bind-marker}
-	 * {@literal <columns> IN :bind-marker}
 	 */
-	private Function<Map<AggregatePath, Column>, Condition> inCondition() {
+	private Condition inCondition(Map<AggregatePath, Column> columnMap) {
-
-		return columnMap -> {
 
-			List<Column> columns = List.copyOf(columnMap.values());
+		List<Column> columns = List.copyOf(columnMap.values());
 
-			if (columns.size() == 1) {
+		if (columns.size() == 1) {
-				return Conditions.in(columns.get(0), getBindMarker(IDS_SQL_PARAMETER));
+			return Conditions.in(columns.get(0), getBindMarker(IDS_SQL_PARAMETER));
-			}
+		}
-			return Conditions.in(TupleExpression.create(columns), getBindMarker(IDS_SQL_PARAMETER));
+		return Conditions.in(TupleExpression.create(columns), getBindMarker(IDS_SQL_PARAMETER));
-		};
 	}
 
 	/**
-	 * Constructs a function for constructing a where. The where condition will be of the form
+	 * Constructs a where-condition. The where condition will be of the form
 	 * {@literal <column-a> = :bind-marker-a AND <column-b> = :bind-marker-b ...}
 	 */
-	private Function<Map<AggregatePath, Column>, Condition> equalityCondition() {
+	private Condition equalityCondition(Map<AggregatePath, Column> columnMap) {
 
 		AggregatePath.ColumnInfos idColumnInfos = mappingContext.getAggregatePath(entity).getTableInfo().idColumnInfos();
 
-		return columnMap -> {
+		Condition result = null;
-
+		for (Map.Entry<AggregatePath, Column> entry : columnMap.entrySet()) {
-			Condition result = null;
+			BindMarker bindMarker = getBindMarker(idColumnInfos.get(entry.getKey()).name());
-			for (Map.Entry<AggregatePath, Column> entry : columnMap.entrySet()) {
+			Comparison singleCondition = entry.getValue().isEqualTo(bindMarker);
-				BindMarker bindMarker = getBindMarker(idColumnInfos.get(entry.getKey()).name());
-				Comparison singleCondition = entry.getValue().isEqualTo(bindMarker);
 
-				result = result == null ? singleCondition : result.and(singleCondition);
+			result = result == null ? singleCondition : result.and(singleCondition);
-			}
+		}
-			return result;
+		Assert.state(result != null, "We need at least one condition");
-		};
+		return result;
 	}
 
 	/**
 	 * Constructs a function for constructing where a condition. The where condition will be of the form
 	 * {@literal <column-a> IS NOT NULL AND <column-b> IS NOT NULL ... }
 	 */
-	private Function<Map<AggregatePath, Column>, Condition> isNotNullCondition() {
+	private Condition isNotNullCondition(Map<AggregatePath, Column> columnMap) {
-
-		return columnMap -> {
 
-			Condition result = null;
+		Condition result = null;
-			for (Column column : columnMap.values()) {
+		for (Column column : columnMap.values()) {
-				Condition singleCondition = column.isNotNull();
+			Condition singleCondition = column.isNotNull();
 
-				result = result == null ? singleCondition : result.and(singleCondition);
+			result = result == null ? singleCondition : result.and(singleCondition);
-			}
+		}
-			return result;
+		Assert.state(result != null, "We need at least one condition");
-		};
+		return result;
 	}
 
 	private String createFindOneSql() {
@@ -601,9 +597,13 @@ class SqlGenerator {
 
 	private SelectBuilder.SelectWhere selectBuilder(Collection<SqlIdentifier> keyColumns, Query query) {
 
-		Table table = getTable();
+		return createSelectBuilder(getTable(), ap -> false, keyColumns);
+	}
+
+	private SelectBuilder.SelectWhere createSelectBuilder(Table table, Predicate<AggregatePath> pathFilter,
+			Collection<SqlIdentifier> keyColumns) {
 
-		Projection projection = getProjection(keyColumns, query, table);
+		Projection projection = getProjection(pathFilter, keyColumns, query, table);
 		SelectBuilder.SelectJoin baseSelect = StatementBuilder.select(projection.columns()).from(table);
 
 		return (SelectBuilder.SelectWhere) addJoins(baseSelect, joinTables);
@@ -613,18 +613,12 @@ class SqlGenerator {
 
 		for (Join join : projection.joins()) {
 
-			Condition condition = null;
+			baseSelect = baseSelect.leftOuterJoin(join.joinTable).on(join.condition);
-			for (Pair<Column, Column> columnPair : join.columns) {
-				Comparison elementalCondition = columnPair.getFirst().isEqualTo(columnPair.getSecond());
-				condition = condition == null ? elementalCondition : condition.and(elementalCondition);
-			}
-
-			baseSelect = baseSelect.leftOuterJoin(join.joinTable).on(Objects.requireNonNull(condition));
 		}
 		return baseSelect;
 	}
 
-	private Projection getProjection(Collection<SqlIdentifier> keyColumns, Query query, Table table) {
+	private Projection getProjection(Predicate<AggregatePath> pathFilter, Collection<SqlIdentifier> keyColumns, Query query, Table table) {
 
 		Set<Expression> columns = new LinkedHashSet<>();
 		Set<Join> joins = new LinkedHashSet<>();
@@ -648,6 +642,10 @@ class SqlGenerator {
 
 				AggregatePath aggregatePath = mappingContext.getAggregatePath(path);
 
+				if (pathFilter.test(aggregatePath)) {
+				continue;
+			}
+
 				includeColumnAndJoin(aggregatePath, joins, columns);
 			}
 		}
@@ -762,21 +760,25 @@ class SqlGenerator {
 		}
 
 		Table currentTable = sqlContext.getTable(path);
-		AggregatePath.ColumnInfos backRefColumnInfos = path.getTableInfo().reverseColumnInfos();
+		AggregatePath.ColumnInfos backRefColumnInfos = path.getTableInfo().backReferenceColumnInfos();
 
 		AggregatePath idDefiningParentPath = path.getIdDefiningParentPath();
 		Table parentTable = sqlContext.getTable(idDefiningParentPath);
 		AggregatePath.ColumnInfos idColumnInfos = idDefiningParentPath.getTableInfo().idColumnInfos();
 
-		List<Pair<Column, Column>> joinConditions = new ArrayList<>();
+		final Condition[] joinCondition = { null };
 		backRefColumnInfos.forEach((ap, ci) -> {
-			joinConditions.add(Pair.of(currentTable.column(ci.name()), parentTable.column(idColumnInfos.get(ap).name())));
+
+			Condition elementalCondition = currentTable.column(ci.name())
+					.isEqualTo(parentTable.column(idColumnInfos.get(ap).name()));
+			joinCondition[0] = joinCondition[0] == null ? elementalCondition : joinCondition[0].and(elementalCondition);
 		});
 
 		return new Join( //
 				currentTable, //
-				joinConditions //
+				joinCondition[0] //
 		);
+
 	}
 
 	private String createFindAllInListSql() {
@@ -914,7 +916,7 @@ class SqlGenerator {
 		Delete delete;
 
 		Map<AggregatePath, Column> columns = new TreeMap<>();
-		AggregatePath.ColumnInfos columnInfos = path.getTableInfo().reverseColumnInfos();
+		AggregatePath.ColumnInfos columnInfos = path.getTableInfo().backReferenceColumnInfos();
 		columnInfos.forEach((ag, ci) -> columns.put(ag, table.column(ci.name())));
 
 		if (isFirstNonRoot(path)) {
@@ -1225,7 +1227,7 @@ class SqlGenerator {
 	/**
 	 * Value object representing a {@code JOIN} association.
 	 */
-	record Join(Table joinTable, List<Pair<Column, Column>> columns) {
+	record Join(Table joinTable, Condition condition) {
 	}
 
 	/**

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

@@ -56,7 +56,7 @@ public class SqlGeneratorSource {
 		return dialect;
 	}
 
-	SqlGenerator getSqlGenerator(Class<?> domainType) {
+	public SqlGenerator getSqlGenerator(Class<?> domainType) {
 
 		return CACHE.computeIfAbsent(domainType,
 				t -> new SqlGenerator(context, converter, context.getRequiredPersistentEntity(t), dialect));

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

@@ -19,6 +19,8 @@ import java.sql.SQLType;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.function.BiFunction;
+import java.util.function.Function;
 import java.util.function.Predicate;
 
 import org.springframework.data.jdbc.core.mapping.JdbcValue;
@@ -122,30 +124,18 @@ public class SqlParametersFactory {
 		RelationalPersistentEntity<T> entity = getRequiredPersistentEntity(domainType);
 		RelationalPersistentProperty singleIdProperty = entity.getRequiredIdProperty();
 
-		if (singleIdProperty.isEntity()) {
+		RelationalPersistentEntity<?> complexId = context.getPersistentEntity(singleIdProperty);
 
-			RelationalPersistentEntity<?> complexId = context.getPersistentEntity(singleIdProperty);
+		Function<AggregatePath, Object> valueExtractor = complexId == null ? ap -> id
-			PersistentPropertyPathAccessor<Object> accessor = complexId.getPropertyPathAccessor(id);
+				: ap -> complexId.getPropertyPathAccessor(id).getProperty(ap.getRequiredPersistentPropertyPath());
 
-			context.getAggregatePath(entity).getTableInfo().idColumnInfos().forEach((ap, ci) -> {
+		context.getAggregatePath(entity).getTableInfo().idColumnInfos() //
-				Object idValue = accessor.getProperty(ap.getRequiredPersistentPropertyPath());
+				.forEach((ap, ci) -> addConvertedPropertyValue( //
-
-				addConvertedPropertyValue( //
 						parameterSource, //
 						ap.getRequiredLeafProperty(), //
-						idValue, //
+						valueExtractor.apply(ap), //
 						ci.name() //
-				);
+				));
-			});
-		} else {
-
-			addConvertedPropertyValue( //
-					parameterSource, //
-					singleIdProperty, //
-					id, //
-					singleIdProperty.getColumnName() //
-			);
-		}
 		return parameterSource;
 	}
 
@@ -161,32 +151,26 @@ public class SqlParametersFactory {
 
 		SqlIdentifierParameterSource parameterSource = new SqlIdentifierParameterSource();
 
-		RelationalPersistentEntity<?> entity = context.getPersistentEntity(domainType);
+		RelationalPersistentEntity<?> entity = context.getRequiredPersistentEntity(domainType);
 		RelationalPersistentProperty singleIdProperty = entity.getRequiredIdProperty();
+		RelationalPersistentEntity<?> complexId = context.getPersistentEntity(singleIdProperty);
+		AggregatePath.ColumnInfos idColumnInfos = context.getAggregatePath(entity).getTableInfo().idColumnInfos();
 
-		if (singleIdProperty.isEntity()) {
+		BiFunction<Object, AggregatePath, Object> valueExtractor = complexId == null ? (id, ap) -> id
-
+				: (id, ap) -> complexId.getPropertyPathAccessor(id).getProperty(ap.getRequiredPersistentPropertyPath());
-			RelationalPersistentEntity<?> complexId = context.getPersistentEntity(singleIdProperty);
-
-			AggregatePath.ColumnInfos idColumnInfos = context.getAggregatePath(entity).getTableInfo().idColumnInfos();
 
-			List<Object[]> parameterValues = new ArrayList<>();
+		List<Object[]> parameterValues = new ArrayList<>();
-			for (Object id : ids) {
+		for (Object id : ids) {
 
-				PersistentPropertyPathAccessor<Object> accessor = complexId.getPropertyPathAccessor(id);
+			List<Object> tupleList = new ArrayList<>();
+			idColumnInfos.forEach((ap, ci) -> {
+				tupleList.add(valueExtractor.apply(id, ap));
+			});
+			parameterValues.add(tupleList.toArray(new Object[0]));
+		}
 
-				List<Object> tupleList = new ArrayList<>();
+		parameterSource.addValue(SqlGenerator.IDS_SQL_PARAMETER, parameterValues);
-				idColumnInfos.forEach((ap, ci) -> {
-					tupleList.add(accessor.getProperty(ap.getRequiredPersistentPropertyPath()));
-				});
-				parameterValues.add(tupleList.toArray(new Object[0]));
-			}
 
-			parameterSource.addValue(SqlGenerator.IDS_SQL_PARAMETER, parameterValues);
-		} else {
-			addConvertedPropertyValuesAsList(parameterSource, getRequiredPersistentEntity(domainType).getRequiredIdProperty(),
-					ids);
-		}
 		return parameterSource;
 	}
 

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

@@ -29,9 +29,17 @@ 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.Criteria;
-import org.springframework.data.relational.core.sql.*;
+import org.springframework.data.relational.core.sql.Column;
+import org.springframework.data.relational.core.sql.Condition;
+import org.springframework.data.relational.core.sql.Conditions;
+import org.springframework.data.relational.core.sql.Delete;
 import org.springframework.data.relational.core.sql.DeleteBuilder.DeleteWhere;
+import org.springframework.data.relational.core.sql.Expression;
+import org.springframework.data.relational.core.sql.Expressions;
+import org.springframework.data.relational.core.sql.Select;
 import org.springframework.data.relational.core.sql.SelectBuilder.SelectWhere;
+import org.springframework.data.relational.core.sql.StatementBuilder;
+import org.springframework.data.relational.core.sql.Table;
 import org.springframework.data.relational.core.sql.render.SqlRenderer;
 import org.springframework.data.relational.repository.query.RelationalEntityMetadata;
 import org.springframework.data.relational.repository.query.RelationalParameterAccessor;
@@ -89,13 +97,10 @@ class JdbcDeleteQueryCreator extends RelationalQueryCreator<List<ParametrizedQue
 		Table table = Table.create(entityMetadata.getTableName());
 		MapSqlParameterSource parameterSource = new MapSqlParameterSource();
 
-		SqlContext sqlContext = new SqlContext();
-
 		Condition condition = criteria == null ? null
 				: queryMapper.getMappedObject(parameterSource, criteria, table, entity);
 
-		List<Column> idColumns = context.getAggregatePath(entity).getTableInfo().idColumnInfos()
+		List<Column> idColumns = context.getAggregatePath(entity).getTableInfo().idColumnInfos().toColumnList(table);
-				.toList(ci -> table.column(ci.name()));
 
 		// create select criteria query for subselect
 		SelectWhere selectBuilder = StatementBuilder.select(idColumns).from(table);
@@ -128,26 +133,24 @@ class JdbcDeleteQueryCreator extends RelationalQueryCreator<List<ParametrizedQue
 
 			AggregatePath aggregatePath = context.getAggregatePath(path);
 
-			// prevent duplication on recursive call
+			if (aggregatePath.isEmbedded()) {
-			if (path.getLength() > 1 && !aggregatePath.getParentPath().isEmbedded()) {
 				continue;
 			}
 
-			if (aggregatePath.isEntity() && !aggregatePath.isEmbedded()) {
+			if (aggregatePath.isEntity()) {
 
 				SqlContext sqlContext = new SqlContext();
 
-				// MariaDB prior to 11.6 does not  support aliases for delete statements
+				// MariaDB prior to 11.6 does not support aliases for delete statements
 				Table table = sqlContext.getUnaliasedTable(aggregatePath);
 
-				List<Column> reverseColumns = aggregatePath.getTableInfo().reverseColumnInfos()
+				List<Column> reverseColumns = aggregatePath.getTableInfo().backReferenceColumnInfos().toColumnList(table);
-						.toList(ci -> table.column(ci.name()));
+				Expression expression = Expressions.of(reverseColumns);
-				Expression expression = TupleExpression.maybeWrap(reverseColumns);
 
 				Condition inCondition = Conditions.in(expression, parentSelect);
 
 				List<Column> parentIdColumns = aggregatePath.getIdDefiningParentPath().getTableInfo().idColumnInfos()
-						.toList(ci -> table.column(ci.name()));
+						.toColumnList(table);
 
 				Select select = StatementBuilder.select( //
 						parentIdColumns //

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

@@ -15,14 +15,14 @@
  */
 package org.springframework.data.jdbc.repository.query;
 
-import java.util.ArrayList;
-import java.util.List;
 import java.util.Optional;
+import java.util.function.Predicate;
 
 import org.springframework.data.domain.Pageable;
 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.jdbc.core.convert.SqlGeneratorSource;
 import org.springframework.data.mapping.PersistentPropertyPath;
 import org.springframework.data.relational.core.dialect.Dialect;
 import org.springframework.data.relational.core.dialect.RenderContextFactory;
@@ -31,7 +31,12 @@ 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.Criteria;
-import org.springframework.data.relational.core.sql.*;
+import org.springframework.data.relational.core.sql.Column;
+import org.springframework.data.relational.core.sql.Expressions;
+import org.springframework.data.relational.core.sql.Functions;
+import org.springframework.data.relational.core.sql.Select;
+import org.springframework.data.relational.core.sql.SelectBuilder;
+import org.springframework.data.relational.core.sql.Table;
 import org.springframework.data.relational.core.sql.render.SqlRenderer;
 import org.springframework.data.relational.repository.Lock;
 import org.springframework.data.relational.repository.query.RelationalEntityMetadata;
@@ -65,6 +70,7 @@ class JdbcQueryCreator extends RelationalQueryCreator<ParametrizedQuery> {
 	private final boolean isSliceQuery;
 	private final ReturnedType returnedType;
 	private final Optional<Lock> lockMode;
+	private final SqlGeneratorSource sqlGeneratorSource;
 
 	/**
 	 * Creates new instance of this class with the given {@link PartTree}, {@link JdbcConverter}, {@link Dialect},
@@ -78,16 +84,45 @@ class JdbcQueryCreator extends RelationalQueryCreator<ParametrizedQuery> {
 	 * @param accessor parameter metadata provider, must not be {@literal null}.
 	 * @param isSliceQuery flag denoting if the query returns a {@link org.springframework.data.domain.Slice}.
 	 * @param returnedType the {@link ReturnedType} to be returned by the query. Must not be {@literal null}.
+	 * @deprecated use
+	 *             {@link JdbcQueryCreator#JdbcQueryCreator(RelationalMappingContext, PartTree, JdbcConverter, Dialect, RelationalEntityMetadata, RelationalParameterAccessor, boolean, ReturnedType, Optional, SqlGeneratorSource)}
+	 *             instead.
 	 */
+	@Deprecated(since = "4.0", forRemoval = true)
 	JdbcQueryCreator(RelationalMappingContext context, PartTree tree, JdbcConverter converter, Dialect dialect,
 			RelationalEntityMetadata<?> entityMetadata, RelationalParameterAccessor accessor, boolean isSliceQuery,
 			ReturnedType returnedType, Optional<Lock> lockMode) {
+		this(context, tree, converter, dialect, entityMetadata, accessor, isSliceQuery, returnedType, lockMode,
+				new SqlGeneratorSource(context, converter, dialect));
+	}
+
+	/**
+	 * Creates new instance of this class with the given {@link PartTree}, {@link JdbcConverter}, {@link Dialect},
+	 * {@link RelationalEntityMetadata} and {@link RelationalParameterAccessor}.
+	 *
+	 * @param context the mapping context. Must not be {@literal null}.
+	 * @param tree part tree, must not be {@literal null}.
+	 * @param converter must not be {@literal null}.
+	 * @param dialect must not be {@literal null}.
+	 * @param entityMetadata relational entity metadata, must not be {@literal null}.
+	 * @param accessor parameter metadata provider, must not be {@literal null}.
+	 * @param isSliceQuery flag denoting if the query returns a {@link org.springframework.data.domain.Slice}.
+	 * @param returnedType the {@link ReturnedType} to be returned by the query. Must not be {@literal null}.
+	 * @param lockMode lock mode to be used for the query.
+	 * @param sqlGeneratorSource the source providing SqlGenerator instances for generating SQL. Must not be
+	 *          {@literal null}
+	 * @since 4.0
+	 */
+	JdbcQueryCreator(RelationalMappingContext context, PartTree tree, JdbcConverter converter, Dialect dialect,
+			RelationalEntityMetadata<?> entityMetadata, RelationalParameterAccessor accessor, boolean isSliceQuery,
+			ReturnedType returnedType, Optional<Lock> lockMode, SqlGeneratorSource sqlGeneratorSource) {
 		super(tree, accessor);
 
 		Assert.notNull(converter, "JdbcConverter must not be null");
 		Assert.notNull(dialect, "Dialect must not be null");
 		Assert.notNull(entityMetadata, "Relational entity metadata must not be null");
 		Assert.notNull(returnedType, "ReturnedType must not be null");
+		Assert.notNull(sqlGeneratorSource, "SqlGeneratorSource must not be null");
 
 		this.context = context;
 		this.tree = tree;
@@ -99,6 +134,7 @@ class JdbcQueryCreator extends RelationalQueryCreator<ParametrizedQuery> {
 		this.isSliceQuery = isSliceQuery;
 		this.returnedType = returnedType;
 		this.lockMode = lockMode;
+		this.sqlGeneratorSource = sqlGeneratorSource;
 	}
 
 	/**
@@ -228,122 +264,13 @@ class JdbcQueryCreator extends RelationalQueryCreator<ParametrizedQuery> {
 
 	private SelectBuilder.SelectJoin selectBuilder(Table table) {
 
-		List<Expression> columnExpressions = new ArrayList<>();
 		RelationalPersistentEntity<?> entity = entityMetadata.getTableEntity();
-		SqlContext sqlContext = new SqlContext();
-
-		List<Join> joinTables = new ArrayList<>();
-		for (PersistentPropertyPath<RelationalPersistentProperty> path : context
-				.findPersistentPropertyPaths(entity.getType(), p -> true)) {
-
-			AggregatePath aggregatePath = context.getAggregatePath(path);
-
-			if (returnedType.needsCustomConstruction()) {
-				if (!returnedType.getInputProperties().contains(aggregatePath.getRequiredBaseProperty().getName())) {
-					continue;
-				}
-			}
-
-			// add a join if necessary
-			Join join = getJoin(sqlContext, aggregatePath);
-			if (join != null) {
-				joinTables.add(join);
-			}
-
-			Column column = getColumn(sqlContext, aggregatePath);
-			if (column != null) {
-				columnExpressions.add(column);
-			}
-		}
-
-		SelectBuilder.SelectAndFrom selectBuilder = StatementBuilder.select(columnExpressions);
-		SelectBuilder.SelectJoin baseSelect = selectBuilder.from(table);
-
-		for (Join join : joinTables) {
-
-			Condition condition = null;
 
-			for (int i = 0; i < join.joinColumns.size(); i++) {
+		Predicate<AggregatePath> filter = ap -> returnedType.needsCustomConstruction()
-				Column parentColumn = join.parentId.get(i);
+				&& !returnedType.getInputProperties().contains(ap.getRequiredBaseProperty().getName());
-				Column joinColumn = join.joinColumns.get(i);
-				Comparison singleCondition = joinColumn.isEqualTo(parentColumn);
-				condition = condition == null ? singleCondition : condition.and(singleCondition);
-			}
-
-			Assert.state(condition != null, "No condition found");
-
-			baseSelect = baseSelect.leftOuterJoin(join.joinTable).on(condition);
-		}
 
-		return baseSelect;
+		return (SelectBuilder.SelectJoin) sqlGeneratorSource.getSqlGenerator(entity.getType()).createSelectBuilder(table,
+				filter);
 	}
 
-	/**
-	 * Create a {@link Column} for {@link AggregatePath}.
-	 *
-	 * @param sqlContext for generating SQL constructs.
-	 * @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, 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
-		// select
-		// only the parent path is considered in order to handle arrays that get stored as BINARY properly
-		if (path.isEmbedded() || path.getParentPath().isMultiValued()) {
-			return null;
-		}
-
-		if (path.isEntity()) {
-
-			if (path.isQualified() //
-					|| path.isCollectionLike() //
-					|| path.hasIdProperty() //
-			) {
-				return null;
-			}
-
-			// Simple entities without id include there backreference as an synthetic id in order to distinguish null entities
-			// from entities with only null values.
-			return sqlContext.getAnyReverseColumn(path);
-		}
-
-		return sqlContext.getColumn(path);
-	}
-
-	@Nullable
-	Join getJoin(SqlContext sqlContext, AggregatePath path) {
-
-		if (!path.isEntity() || path.isEmbedded() || path.isMultiValued()) {
-			return null;
-		}
-
-		Table currentTable = sqlContext.getTable(path);
-
-		AggregatePath idDefiningParentPath = path.getIdDefiningParentPath();
-		Table parentTable = sqlContext.getTable(idDefiningParentPath);
-
-		List<Column> reverseColumns = path.getTableInfo().reverseColumnInfos().toList(ci -> currentTable.column(ci.name()));
-		List<Column> idColumns = idDefiningParentPath.getTableInfo().idColumnInfos()
-				.toList(ci -> parentTable.column(ci.name()));
-		return new Join( //
-				currentTable, //
-				reverseColumns, //
-				idColumns //
-		);
-	}
-
-	/**
-	 * Value object representing a {@code JOIN} association.
-	 */
-	private record Join(Table joinTable, List<Column> joinColumns, List<Column> parentId) {
-
-		Join {
-			Assert.isTrue(joinColumns.size() == parentId.size(),
-					"Both sides of a join condition must have the same number of columns");
-		}
-
-	}
 }

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

@@ -47,7 +47,7 @@ class SqlContext {
 
 	Column getAnyReverseColumn(AggregatePath path) {
 
-		AggregatePath.ColumnInfo anyReverseColumnInfo = path.getTableInfo().reverseColumnInfos().any();
+		AggregatePath.ColumnInfo anyReverseColumnInfo = path.getTableInfo().backReferenceColumnInfos().any();
 		return getTable(path).column(anyReverseColumnInfo.name()).as(anyReverseColumnInfo.alias());
 	}
 

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

@@ -18,9 +18,7 @@ package org.springframework.data.jdbc.core;
 import static org.assertj.core.api.Assertions.*;
 
 import java.util.List;
-import java.util.Optional;
 
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationEventPublisher;
@@ -244,8 +242,9 @@ public class CompositeIdAggregateTemplateHsqlIntegrationTests {
 						new EmbeddedPk(23L, "x"), "alpha" //
 				));
 
-		Query projectingQuery = Query.empty().columns( "embeddedPk.two", "name");
+		Query projectingQuery = Query.empty().columns("embeddedPk.two", "name");
-		SimpleEntityWithEmbeddedPk projected = template.findOne(projectingQuery, SimpleEntityWithEmbeddedPk.class).orElseThrow();
+		SimpleEntityWithEmbeddedPk projected = template.findOne(projectingQuery, SimpleEntityWithEmbeddedPk.class)
+				.orElseThrow();
 
 		// Projection still does a full select, otherwise one would be null.
 		// See https://github.com/spring-projects/spring-data-relational/issues/1821

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

@@ -120,7 +120,8 @@ public class JdbcAggregateChangeExecutorContextImmutableUnitTests {
 		assertThat(newRoot.list.get(0).id).isEqualTo(24L);
 	}
 
-	@Test // GH-537
+	@Test
+	// GH-537
 	void populatesIdsIfNecessaryForAllRootsThatWereProcessed() {
 
 		DummyEntity root1 = new DummyEntity().withId(123L);
@@ -166,7 +167,8 @@ public class JdbcAggregateChangeExecutorContextImmutableUnitTests {
 	}
 
 	Identifier createBackRef(long value) {
-		return JdbcIdentifierBuilder.forBackReferences(converter, toAggregatePath("content"), value).build();
+		return JdbcIdentifierBuilder.forBackReferences(converter, toAggregatePath("content"),
+				JdbcAggregateChangeExecutionContext.getValueProvider(value, toAggregatePath("content"), converter)).build();
 	}
 
 	PersistentPropertyPath<RelationalPersistentProperty> toPath(String path) {
@@ -180,10 +182,8 @@ public class JdbcAggregateChangeExecutorContextImmutableUnitTests {
 
 	private static final class DummyEntity {
 
-		@Id
+		@Id private final Long id;
-		private final Long id;
+		@Version private final long version;
-		@Version
-		private final long version;
 
 		private final Content content;
 
@@ -221,14 +221,16 @@ public class JdbcAggregateChangeExecutorContextImmutableUnitTests {
 		}
 
 		public boolean equals(final Object o) {
-			if (o == this) return true;
+			if (o == this)
+				return true;
 			if (!(o instanceof final DummyEntity other))
 				return false;
 			final Object this$id = this.getId();
 			final Object other$id = other.getId();
 			if (!Objects.equals(this$id, other$id))
 				return false;
-			if (this.getVersion() != other.getVersion()) return false;
+			if (this.getVersion() != other.getVersion())
+				return false;
 			final Object this$content = this.getContent();
 			final Object other$content = other.getContent();
 			if (!Objects.equals(this$content, other$content))
@@ -253,7 +255,8 @@ public class JdbcAggregateChangeExecutorContextImmutableUnitTests {
 		}
 
 		public String toString() {
-			return "JdbcAggregateChangeExecutorContextImmutableUnitTests.DummyEntity(id=" + this.getId() + ", version=" + this.getVersion() + ", content=" + this.getContent() + ", list=" + this.getList() + ")";
+			return "JdbcAggregateChangeExecutorContextImmutableUnitTests.DummyEntity(id=" + this.getId() + ", version="
+					+ this.getVersion() + ", content=" + this.getContent() + ", list=" + this.getList() + ")";
 		}
 
 		public DummyEntity withId(Long id) {
@@ -274,8 +277,7 @@ public class JdbcAggregateChangeExecutorContextImmutableUnitTests {
 	}
 
 	private static final class Content {
-		@Id
+		@Id private final Long id;
-		private final Long id;
 
 		Content() {
 			id = null;
@@ -290,7 +292,8 @@ public class JdbcAggregateChangeExecutorContextImmutableUnitTests {
 		}
 
 		public boolean equals(final Object o) {
-			if (o == this) return true;
+			if (o == this)
+				return true;
 			if (!(o instanceof final Content other))
 				return false;
 			final Object this$id = this.getId();

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

@@ -18,6 +18,7 @@ package org.springframework.data.jdbc.core;
 import static java.util.Collections.*;
 import static org.assertj.core.api.Assertions.*;
 import static org.mockito.Mockito.*;
+import static org.springframework.data.jdbc.core.JdbcAggregateChangeExecutionContext.*;
 import static org.springframework.data.jdbc.core.convert.JdbcIdentifierBuilder.*;
 
 import java.util.ArrayList;
@@ -257,7 +258,8 @@ public class JdbcAggregateChangeExecutorContextUnitTests {
 	}
 
 	Identifier createBackRef(long value) {
-		return forBackReferences(converter, toAggregatePath("content"), value).build();
+		return forBackReferences(converter, toAggregatePath("content"),
+				getValueProvider(value, toAggregatePath("content"), converter)).build();
 	}
 
 	PersistentPropertyPath<RelationalPersistentProperty> toPath(String path) {

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

@@ -21,14 +21,17 @@ import static org.springframework.data.relational.core.sql.SqlIdentifier.*;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
+import java.util.function.Function;
 
 import org.junit.jupiter.api.Nested;
 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.mapping.PersistentPropertyPathAccessor;
 import org.springframework.data.relational.core.mapping.AggregatePath;
 import org.springframework.data.relational.core.mapping.Embedded;
+import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 
 /**
  * Unit tests for the {@link JdbcIdentifierBuilder}.
@@ -47,7 +50,9 @@ public class JdbcIdentifierBuilderUnitTests {
 		@Test // DATAJDBC-326
 		void parametersWithPropertyKeysUseTheParentPropertyJdbcType() {
 
-			Identifier identifier = JdbcIdentifierBuilder.forBackReferences(converter, getPath("child"), "eins").build();
+			Identifier identifier = JdbcIdentifierBuilder
+					.forBackReferences(converter, getPath("child"), getValueProvider("eins", getPath("child"), converter))
+					.build();
 
 			assertThat(identifier.getParts()) //
 					.extracting("name", "value", "targetType") //
@@ -62,7 +67,7 @@ public class JdbcIdentifierBuilderUnitTests {
 			AggregatePath path = getPath("children");
 
 			Identifier identifier = JdbcIdentifierBuilder //
-					.forBackReferences(converter, path, "parent-eins") //
+					.forBackReferences(converter, path, getValueProvider("parent-eins", path, converter)) //
 					.withQualifier(path, "map-key-eins") //
 					.build();
 
@@ -80,7 +85,7 @@ public class JdbcIdentifierBuilderUnitTests {
 			AggregatePath path = getPath("moreChildren");
 
 			Identifier identifier = JdbcIdentifierBuilder //
-					.forBackReferences(converter, path, "parent-eins") //
+					.forBackReferences(converter, path, getValueProvider("parent-eins", path, converter)) //
 					.withQualifier(path, "list-index-eins") //
 					.build();
 
@@ -96,7 +101,8 @@ public class JdbcIdentifierBuilderUnitTests {
 		void backreferenceAcrossEmbeddable() {
 
 			Identifier identifier = JdbcIdentifierBuilder //
-					.forBackReferences(converter, getPath("embeddable.child"), "parent-eins") //
+					.forBackReferences(converter, getPath("embeddable.child"),
+							getValueProvider("parent-eins", getPath("embeddable.child"), converter)) //
 					.build();
 
 			assertThat(identifier.getParts()) //
@@ -110,7 +116,8 @@ public class JdbcIdentifierBuilderUnitTests {
 		void backreferenceAcrossNoId() {
 
 			Identifier identifier = JdbcIdentifierBuilder //
-					.forBackReferences(converter, getPath("noId.child"), "parent-eins") //
+					.forBackReferences(converter, getPath("noId.child"),
+							getValueProvider("parent-eins", getPath("noId.child"), converter)) //
 					.build();
 
 			assertThat(identifier.getParts()) //
@@ -125,6 +132,25 @@ public class JdbcIdentifierBuilderUnitTests {
 		}
 	}
 
+	/**
+	 * copied from JdbcAggregateChangeExecutionContext
+	 */
+	static Function<AggregatePath, Object> getValueProvider(Object idValue, AggregatePath path, JdbcConverter converter) {
+
+		RelationalPersistentEntity<?> entity = converter.getMappingContext()
+				.getPersistentEntity(path.getIdDefiningParentPath().getRequiredIdProperty().getType());
+
+		Function<AggregatePath, Object> valueProvider = ap -> {
+			if (entity == null) {
+				return idValue;
+			} else {
+				PersistentPropertyPathAccessor<Object> propertyPathAccessor = entity.getPropertyPathAccessor(idValue);
+				return propertyPathAccessor.getProperty(ap.getRequiredPersistentPropertyPath());
+			}
+		};
+		return valueProvider;
+	}
+
 	@Nested
 	class WithCompositeId {
 
@@ -136,7 +162,7 @@ public class JdbcIdentifierBuilderUnitTests {
 			AggregatePath path = getPath("children");
 
 			Identifier identifier = JdbcIdentifierBuilder //
-					.forBackReferences(converter, path, exampleId) //
+					.forBackReferences(converter, path, getValueProvider(exampleId, path, converter)) //
 					.build();
 
 			assertThat(identifier.getParts()) //

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

@@ -89,42 +89,42 @@ class SqlGeneratorEmbeddedUnitTests {
 	@Test // GH-574
 	void findOneWrappedId() {
 
-		SqlGenerator sqlGenerator = createSqlGenerator(DummyEntityWithWrappedId.class);
+		SqlGenerator sqlGenerator = createSqlGenerator(WithWrappedId.class);
 
 		String sql = sqlGenerator.getFindOne();
 
 		assertSoftly(softly -> {
 
 			softly.assertThat(sql).startsWith("SELECT") //
-					.contains("dummy_entity_with_wrapped_id.name AS name") //
+					.contains("with_wrapped_id.name AS name") //
-					.contains("dummy_entity_with_wrapped_id.id") //
+					.contains("with_wrapped_id.id") //
-					.contains("WHERE dummy_entity_with_wrapped_id.id = :id");
+					.contains("WHERE with_wrapped_id.id = :id");
 		});
 	}
 
 	@Test // GH-574
 	void findOneEmbeddedId() {
 
-		SqlGenerator sqlGenerator = createSqlGenerator(DummyEntityWithEmbeddedId.class);
+		SqlGenerator sqlGenerator = createSqlGenerator(WithEmbeddedId.class);
 
 		String sql = sqlGenerator.getFindOne();
 
 		assertSoftly(softly -> {
 
 			softly.assertThat(sql).startsWith("SELECT") //
-					.contains("dummy_entity_with_embedded_id.name AS name") //
+					.contains("with_embedded_id.name AS name") //
-					.contains("dummy_entity_with_embedded_id.one") //
+					.contains("with_embedded_id.one") //
-					.contains("dummy_entity_with_embedded_id.two") //
+					.contains("with_embedded_id.two") //
 					.contains(" WHERE ") //
-					.contains("dummy_entity_with_embedded_id.one = :one") //
+					.contains("with_embedded_id.one = :one") //
-					.contains("dummy_entity_with_embedded_id.two = :two");
+					.contains("with_embedded_id.two = :two");
 		});
 	}
 
 	@Test // GH-574
 	void deleteByIdEmbeddedId() {
 
-		SqlGenerator sqlGenerator = createSqlGenerator(DummyEntityWithEmbeddedId.class);
+		SqlGenerator sqlGenerator = createSqlGenerator(WithEmbeddedId.class);
 
 		String sql = sqlGenerator.getDeleteById();
 
@@ -132,15 +132,15 @@ class SqlGeneratorEmbeddedUnitTests {
 
 			softly.assertThat(sql).startsWith("DELETE") //
 					.contains(" WHERE ") //
-					.contains("dummy_entity_with_embedded_id.one = :one") //
+					.contains("with_embedded_id.one = :one") //
-					.contains("dummy_entity_with_embedded_id.two = :two");
+					.contains("with_embedded_id.two = :two");
 		});
 	}
 
 	@Test // GH-574
 	void deleteByIdInEmbeddedId() {
 
-		SqlGenerator sqlGenerator = createSqlGenerator(DummyEntityWithEmbeddedId.class);
+		SqlGenerator sqlGenerator = createSqlGenerator(WithEmbeddedId.class);
 
 		String sql = sqlGenerator.getDeleteByIdIn();
 
@@ -148,33 +148,33 @@ class SqlGeneratorEmbeddedUnitTests {
 
 			softly.assertThat(sql).startsWith("DELETE") //
 					.contains(" WHERE ") //
-					.contains("(dummy_entity_with_embedded_id.one, dummy_entity_with_embedded_id.two) IN (:ids)");
+					.contains("(with_embedded_id.one, with_embedded_id.two) IN (:ids)");
 		});
 	}
 
 	@Test // GH-574
 	void deleteByPathEmbeddedId() {
 
-		SqlGenerator sqlGenerator = createSqlGenerator(DummyEntityWithEmbeddedId.class);
+		SqlGenerator sqlGenerator = createSqlGenerator(WithEmbeddedId.class);
 		PersistentPropertyPath<RelationalPersistentProperty> path = PersistentPropertyPathTestUtils.getPath("other",
-				DummyEntityWithEmbeddedIdAndReference.class, context);
+				WithEmbeddedIdAndReference.class, context);
 
 		String sql = sqlGenerator.createDeleteByPath(path);
 
 		assertSoftly(softly -> {
 
 			softly.assertThat(sql).startsWith("DELETE FROM other_entity WHERE") //
-					.contains("other_entity.dummy_entity_with_embedded_id_and_reference_one = :one") //
+					.contains("other_entity.with_embedded_id_and_reference_one = :one") //
-					.contains("other_entity.dummy_entity_with_embedded_id_and_reference_two = :two");
+					.contains("other_entity.with_embedded_id_and_reference_two = :two");
 		});
 	}
 
 	@Test // GH-574
 	void deleteInByPathEmbeddedId() {
 
-		SqlGenerator sqlGenerator = createSqlGenerator(DummyEntityWithEmbeddedId.class);
+		SqlGenerator sqlGenerator = createSqlGenerator(WithEmbeddedId.class);
 		PersistentPropertyPath<RelationalPersistentProperty> path = PersistentPropertyPathTestUtils.getPath("other",
-				DummyEntityWithEmbeddedIdAndReference.class, context);
+				WithEmbeddedIdAndReference.class, context);
 
 		String sql = sqlGenerator.createDeleteInByPath(path);
 
@@ -183,14 +183,14 @@ class SqlGeneratorEmbeddedUnitTests {
 			softly.assertThat(sql).startsWith("DELETE FROM other_entity WHERE") //
 					.contains(" WHERE ") //
 					.contains(
-							"(other_entity.dummy_entity_with_embedded_id_and_reference_one, other_entity.dummy_entity_with_embedded_id_and_reference_two) IN (:ids)");
+							"(other_entity.with_embedded_id_and_reference_one, other_entity.with_embedded_id_and_reference_two) IN (:ids)");
 		});
 	}
 
 	@Test // GH-574
 	void updateWithEmbeddedId() {
 
-		SqlGenerator sqlGenerator = createSqlGenerator(DummyEntityWithEmbeddedId.class);
+		SqlGenerator sqlGenerator = createSqlGenerator(WithEmbeddedId.class);
 
 		String sql = sqlGenerator.getUpdate();
 
@@ -198,15 +198,15 @@ class SqlGeneratorEmbeddedUnitTests {
 
 			softly.assertThat(sql).startsWith("UPDATE") //
 					.contains(" WHERE ") //
-					.contains("dummy_entity_with_embedded_id.one = :one") //
+					.contains("with_embedded_id.one = :one") //
-					.contains("dummy_entity_with_embedded_id.two = :two");
+					.contains("with_embedded_id.two = :two");
 		});
 	}
 
 	@Test // GH-574
 	void existsByIdEmbeddedId() {
 
-		SqlGenerator sqlGenerator = createSqlGenerator(DummyEntityWithEmbeddedId.class);
+		SqlGenerator sqlGenerator = createSqlGenerator(WithEmbeddedId.class);
 
 		String sql = sqlGenerator.getExists();
 
@@ -214,8 +214,8 @@ class SqlGeneratorEmbeddedUnitTests {
 
 			softly.assertThat(sql).startsWith("SELECT COUNT") //
 					.contains(" WHERE ") //
-					.contains("dummy_entity_with_embedded_id.one = :one") //
+					.contains("with_embedded_id.one = :one") //
-					.contains("dummy_entity_with_embedded_id.two = :two");
+					.contains("with_embedded_id.two = :two");
 		});
 	}
 
@@ -269,24 +269,24 @@ class SqlGeneratorEmbeddedUnitTests {
 	@Test // GH-574
 	void findAllInListEmbeddedId() {
 
-		SqlGenerator sqlGenerator = createSqlGenerator(DummyEntityWithEmbeddedId.class);
+		SqlGenerator sqlGenerator = createSqlGenerator(WithEmbeddedId.class);
 
 		String sql = sqlGenerator.getFindAllInList();
 
 		assertSoftly(softly -> {
 
 			softly.assertThat(sql).startsWith("SELECT") //
-					.contains("dummy_entity_with_embedded_id.name AS name") //
+					.contains("with_embedded_id.name AS name") //
-					.contains("dummy_entity_with_embedded_id.one") //
+					.contains("with_embedded_id.one") //
-					.contains("dummy_entity_with_embedded_id.two") //
+					.contains("with_embedded_id.two") //
-					.contains(" WHERE (dummy_entity_with_embedded_id.one, dummy_entity_with_embedded_id.two) IN (:ids)");
+					.contains(" WHERE (with_embedded_id.one, with_embedded_id.two) IN (:ids)");
 		});
 	}
 
 	@Test // GH-574
 	void findOneWithReference() {
 
-		SqlGenerator sqlGenerator = createSqlGenerator(DummyEntityWithEmbeddedIdAndReference.class);
+		SqlGenerator sqlGenerator = createSqlGenerator(WithEmbeddedIdAndReference.class);
 
 		String sql = sqlGenerator.getFindOne();
 
@@ -295,13 +295,11 @@ class SqlGeneratorEmbeddedUnitTests {
 			softly.assertThat(sql).startsWith("SELECT") //
 					.contains(" LEFT OUTER JOIN other_entity other ") //
 					.contains(" ON ") //
-					.contains(
+					.contains(" other.with_embedded_id_and_reference_one = with_embedded_id_and_reference.one ") //
-							" other.dummy_entity_with_embedded_id_and_reference_one = dummy_entity_with_embedded_id_and_reference.one ") //
+					.contains(" other.with_embedded_id_and_reference_two = with_embedded_id_and_reference.two ") //
-					.contains(
-							" other.dummy_entity_with_embedded_id_and_reference_two = dummy_entity_with_embedded_id_and_reference.two ") //
 					.contains(" WHERE ") //
-					.contains("dummy_entity_with_embedded_id_and_reference.one = :one") //
+					.contains("with_embedded_id_and_reference.one = :one") //
-					.contains("dummy_entity_with_embedded_id_and_reference.two = :two");
+					.contains("with_embedded_id_and_reference.two = :two");
 		});
 	}
 
@@ -445,17 +443,8 @@ class SqlGeneratorEmbeddedUnitTests {
 
 		assertSoftly(softly -> {
 			softly.assertThat(join.joinTable().getName()).isEqualTo(SqlIdentifier.unquoted("other_entity"));
-			softly.assertThat(join.columns()).extracting( //
+			softly.assertThat(join.condition())
-					pair -> pair.getFirst().getTable(), //
+					.isEqualTo(SqlGeneratorUnitTests.equalsCondition("dummy_entity2", "id", join.joinTable(), "dummy_entity2"));
-					pair -> pair.getFirst().getName(), //
-					pair -> pair.getSecond().getTable().getName(), //
-					pair -> pair.getSecond().getName() //
-			).contains(tuple( //
-					join.joinTable(), //
-					SqlIdentifier.unquoted("dummy_entity2"), //
-					SqlIdentifier.unquoted("dummy_entity2"), //
-					SqlIdentifier.unquoted("id") //
-			));
 		});
 	}
 
@@ -511,7 +500,7 @@ class SqlGeneratorEmbeddedUnitTests {
 	record WrappedId(Long id) {
 	}
 
-	static class DummyEntityWithWrappedId {
+	static class WithWrappedId {
 
 		@Id
 		@Embedded(onEmpty = OnEmpty.USE_NULL) WrappedId wrappedId;
@@ -522,7 +511,7 @@ class SqlGeneratorEmbeddedUnitTests {
 	record EmbeddedId(Long one, String two) {
 	}
 
-	static class DummyEntityWithEmbeddedId {
+	static class WithEmbeddedId {
 
 		@Id
 		@Embedded(onEmpty = OnEmpty.USE_NULL) EmbeddedId embeddedId;
@@ -531,7 +520,7 @@ class SqlGeneratorEmbeddedUnitTests {
 
 	}
 
-	static class DummyEntityWithEmbeddedIdAndReference {
+	static class WithEmbeddedIdAndReference {
 
 		@Id
 		@Embedded(onEmpty = OnEmpty.USE_NULL) EmbeddedId embeddedId;

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

@@ -57,6 +57,7 @@ import org.springframework.data.relational.core.mapping.RelationalPersistentProp
 import org.springframework.data.relational.core.query.Criteria;
 import org.springframework.data.relational.core.query.Query;
 import org.springframework.data.relational.core.sql.Aliased;
+import org.springframework.data.relational.core.sql.Comparison;
 import org.springframework.data.relational.core.sql.LockMode;
 import org.springframework.data.relational.core.sql.SqlIdentifier;
 import org.springframework.data.relational.core.sql.Table;
@@ -92,6 +93,22 @@ class SqlGeneratorUnitTests {
 	});
 	private SqlGenerator sqlGenerator;
 
+	static Comparison equalsCondition(Table parentTable, SqlIdentifier parentId, Table joinedTable,
+			SqlIdentifier joinedColumn) {
+		return org.springframework.data.relational.core.sql.Column.create(joinedColumn, joinedTable)
+				.isEqualTo(org.springframework.data.relational.core.sql.Column.create(parentId, parentTable));
+	}
+
+	static Comparison equalsCondition(SqlIdentifier parentTable, SqlIdentifier parentId, Table joinedTable,
+			SqlIdentifier joinedColumn) {
+		return equalsCondition(Table.create(parentTable), parentId, joinedTable, joinedColumn);
+	}
+
+	static Comparison equalsCondition(String parentTable, String parentId, Table joinedTable, String joinedColumn) {
+		return equalsCondition(SqlIdentifier.unquoted(parentTable), SqlIdentifier.unquoted(parentId), joinedTable,
+				SqlIdentifier.unquoted(joinedColumn));
+	}
+
 	@BeforeEach
 	void setUp() {
 		this.sqlGenerator = createSqlGenerator(DummyEntity.class);
@@ -763,17 +780,9 @@ class SqlGeneratorUnitTests {
 		assertSoftly(softly -> {
 
 			softly.assertThat(join.joinTable().getName()).isEqualTo(SqlIdentifier.quoted("REFERENCED_ENTITY"));
-			softly.assertThat(join.columns()).extracting( //
+			softly.assertThat(join.condition()).isEqualTo(equalsCondition(SqlIdentifier.quoted("DUMMY_ENTITY"),
-					pair -> pair.getFirst().getTable(), //
+					SqlIdentifier.quoted("id1"), join.joinTable(), SqlIdentifier.quoted("DUMMY_ENTITY")));
-					pair -> pair.getFirst().getName(), //
+
-					pair -> pair.getSecond().getTable().getName(), //
-					pair -> pair.getSecond().getName() //
-			).contains(tuple( //
-					join.joinTable(), //
-					SqlIdentifier.quoted("DUMMY_ENTITY"), //
-					SqlIdentifier.quoted("DUMMY_ENTITY"), //
-					SqlIdentifier.quoted("id1") //
-			));
 		});
 	}
 
@@ -801,17 +810,10 @@ class SqlGeneratorUnitTests {
 
 		assertSoftly(softly -> {
 			softly.assertThat(join.joinTable().getName()).isEqualTo(SqlIdentifier.quoted("SECOND_LEVEL_REFERENCED_ENTITY"));
-			softly.assertThat(join.columns()).extracting( //
+			softly.assertThat(join.condition())
-					pair -> pair.getFirst().getTable(), //
+					.isEqualTo(equalsCondition(Table.create("REFERENCED_ENTITY").as(SqlIdentifier.quoted("ref")),
-					pair -> pair.getFirst().getName(), //
+							SqlIdentifier.quoted("X_L1ID"), join.joinTable(), SqlIdentifier.quoted("REFERENCED_ENTITY")));
-					pair -> pair.getSecond().getTable().getName(), //
+
-					pair -> pair.getSecond().getName() //
-			).contains(tuple( //
-					join.joinTable(), //
-					SqlIdentifier.quoted("REFERENCED_ENTITY"), //
-					SqlIdentifier.quoted("REFERENCED_ENTITY"), //
-					SqlIdentifier.quoted("X_L1ID") //
-			));
 		});
 	}
 
@@ -826,18 +828,8 @@ class SqlGeneratorUnitTests {
 			softly.assertThat(joinTable.getName()).isEqualTo(SqlIdentifier.quoted("NO_ID_CHILD"));
 			softly.assertThat(joinTable).isInstanceOf(Aliased.class);
 			softly.assertThat(((Aliased) joinTable).getAlias()).isEqualTo(SqlIdentifier.quoted("child"));
-
+			softly.assertThat(join.condition()).isEqualTo(equalsCondition(SqlIdentifier.quoted("PARENT_OF_NO_ID_CHILD"),
-			softly.assertThat(join.columns()).extracting( //
+					SqlIdentifier.quoted("X_ID"), join.joinTable(), SqlIdentifier.quoted("PARENT_OF_NO_ID_CHILD")));
-					pair -> pair.getFirst().getTable(), //
-					pair -> pair.getFirst().getName(), //
-					pair -> pair.getSecond().getTable().getName(), //
-					pair -> pair.getSecond().getName() //
-			).contains(tuple( //
-					join.joinTable(), //
-					SqlIdentifier.quoted("PARENT_OF_NO_ID_CHILD"), //
-					SqlIdentifier.quoted("PARENT_OF_NO_ID_CHILD"), //
-					SqlIdentifier.quoted("X_ID") //
-			));
 		});
 	}
 

spring-data-r2dbc/src/main/java/org/springframework/data/r2dbc/repository/support/SimpleR2dbcRepository.java

@@ -377,7 +377,8 @@ public class SimpleR2dbcRepository<T, ID> implements R2dbcRepository<T, ID> {
 			idEntity.doWithProperties(new PropertyHandler<RelationalPersistentProperty>() {
 				@Override
 				public void doWithPersistentProperty(RelationalPersistentProperty persistentProperty) {
-					criteriaHolder[0] = criteriaHolder [0].and(persistentProperty.getName()).is(accessor.getProperty(persistentProperty));
+					criteriaHolder[0] = criteriaHolder[0].and(persistentProperty.getName())
+							.is(accessor.getProperty(persistentProperty));
 				}
 			});
 			criteria = criteriaHolder[0];

spring-data-r2dbc/src/test/java/org/springframework/data/r2dbc/repository/CompositeIdRepositoryIntegrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2025 the original author or authors.
+ * Copyright 2025 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.
@@ -15,7 +15,10 @@
  */
 package org.springframework.data.r2dbc.repository;
 
+import static org.assertj.core.api.Assertions.*;
+
 import io.r2dbc.spi.ConnectionFactory;
+import reactor.test.StepVerifier;
 
 import javax.sql.DataSource;
 
@@ -36,9 +39,6 @@ import org.springframework.data.relational.core.mapping.Table;
 import org.springframework.data.repository.reactive.ReactiveCrudRepository;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.test.context.junit.jupiter.SpringExtension;
-import reactor.test.StepVerifier;
-
-import static org.assertj.core.api.Assertions.*;
 
 /**
  * Integration tests for repositories of entities with a composite id.
@@ -103,7 +103,7 @@ public class CompositeIdRepositoryIntegrationTests {
 	void findAllById() {
 		repository.findById(new CompositeId(42, "HBAR")) //
 				.as(StepVerifier::create) //
-				.consumeNextWith(actual ->{
+				.consumeNextWith(actual -> {
 					assertThat(actual.name).isEqualTo("Walter");
 				}).verifyComplete();
 	}

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

@@ -20,11 +20,9 @@ import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.TreeMap;
 import java.util.function.BiConsumer;
 import java.util.function.BiFunction;
 import java.util.function.BinaryOperator;
-import java.util.function.Function;
 import java.util.function.Predicate;
 import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
@@ -32,17 +30,22 @@ import java.util.stream.StreamSupport;
 import org.springframework.data.mapping.PersistentProperty;
 import org.springframework.data.mapping.PersistentPropertyPath;
 import org.springframework.data.mapping.PropertyHandler;
+import org.springframework.data.relational.core.sql.Column;
 import org.springframework.data.relational.core.sql.SqlIdentifier;
+import org.springframework.data.relational.core.sql.Table;
 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.
+ * <p>
+ * It implements {@link Comparable} so that collections of {@code AggregatePath} instances can be sorted in a consistent
+ * way.
  *
- * @since 3.2
  * @author Jens Schauder
  * @author Mark Paluch
+ * @since 3.2
  */
 public interface AggregatePath extends Iterable<AggregatePath>, Comparable<AggregatePath> {
 
@@ -67,7 +70,7 @@ public interface AggregatePath extends Iterable<AggregatePath>, Comparable<Aggre
 	 *
 	 * @param path must not be {@literal null}.
 	 * @return Guaranteed to be not {@literal null}.
-	 * @since 3.5
+	 * @since 4.0
 	 */
 	AggregatePath append(AggregatePath path);
 
@@ -246,39 +249,61 @@ public interface AggregatePath extends Iterable<AggregatePath>, Comparable<Aggre
 	 */
 	AggregatePath getIdDefiningParentPath();
 
+	/**
+	 * The path resulting from removing the first element of the {@link AggregatePath}.
+	 *
+	 * @return {@literal null} for any {@link AggregatePath} having less than two elements.
+	 * @since 4.0
+	 */
 	@Nullable
 	AggregatePath getTail();
 
-	record TableInfo(
+	/**
-
+	 * Subtract the {@literal basePath} from {@literal this} {@literal AggregatePath} by removing the {@literal basePath}
-			/*
+	 * from the beginning of {@literal this}.
-			 * 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.
+	 * @param basePath the path to be removed.
-			 */
+	 * @return an AggregatePath that ends like the original {@literal AggregatePath} but has {@literal basePath} removed
-			SqlIdentifier qualifiedTableName,
+	 *         from the beginning.
-
+	 * @since 4.0
-			/*
+	 */
-			 * The alias used for the table on which this path is based.
+	@Nullable
-			 */
+	AggregatePath subtract(@Nullable AggregatePath basePath);
-			@Nullable SqlIdentifier tableAlias,
-
-			ColumnInfos reverseColumnInfos,
-
-			/*
-			 * 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
+	 * Compares this {@code AggregatePath} to another {@code AggregatePath} based on their dot path notation.
-			 * applicable.
+	 * <p>
-			 */
+	 * This is used to get {@code AggregatePath} instances sorted in a consistent way. Since this order affects generated
-			@Nullable Class<?> qualifierColumnType,
+	 * SQL this also affects query caches and similar.
+	 *
+	 * @param other the {@code AggregatePath} to compare to. Must not be {@literal null}.
+	 * @return a negative integer, zero, or a positive integer as this object's path is less than, equal to, or greater
+	 *         than the specified object's path.
+	 * @since 4.0
+	 */
+	@Override
+	default int compareTo(AggregatePath other) {
+		return toDotPath().compareTo(other.toDotPath());
+	}
 
-			/*
+	/**
-			 * The column name of the id column of the ancestor path that represents an actual table.
+	 * Information about a table underlying an entity.
-			 */
+	 *
-			ColumnInfos idColumnInfos) {
+	 * @param qualifiedTableName 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. Must not be {@literal null}.
+	 * @param tableAlias the alias used for the table on which this path is based. May be {@literal null}.
+	 * @param backReferenceColumnInfos information about the columns used to reference back to the owning entity. Must not
+	 *          be {@literal null}. Since 3.5.
+	 * @param qualifierColumnInfo the column used for the list index or map key of the leaf property of this path. May be
+	 *          {@literal null}.
+	 * @param qualifierColumnType the type of the qualifier column of the leaf property of this path or {@literal null} if
+	 *          this is not applicable. May be {@literal null}.
+	 * @param idColumnInfos the column name of the id column of the ancestor path that represents an actual table. Must
+	 *          not be {@literal null}.
+	 */
+	record TableInfo(SqlIdentifier qualifiedTableName, @Nullable SqlIdentifier tableAlias,
+			ColumnInfos backReferenceColumnInfos, @Nullable ColumnInfo qualifierColumnInfo,
+			@Nullable Class<?> qualifierColumnType, ColumnInfos idColumnInfos) {
 
 		static TableInfo of(AggregatePath path) {
 
@@ -289,7 +314,7 @@ public interface AggregatePath extends Iterable<AggregatePath>, Comparable<Aggre
 
 			SqlIdentifier tableAlias = tableOwner.isRoot() ? null : AggregatePathTableUtils.constructTableAlias(tableOwner);
 
-			ColumnInfos reverseColumnInfos = computeReverseColumnInfo(path);
+			ColumnInfos backReferenceColumnInfos = computeBackReferenceColumnInfos(path);
 
 			ColumnInfo qualifierColumnInfo = null;
 			if (!path.isRoot()) {
@@ -307,8 +332,8 @@ public interface AggregatePath extends Iterable<AggregatePath>, Comparable<Aggre
 
 			ColumnInfos idColumnInfos = computeIdColumnInfos(tableOwner, leafEntity);
 
-			return new TableInfo(qualifiedTableName, tableAlias, reverseColumnInfos, qualifierColumnInfo, qualifierColumnType,
+			return new TableInfo(qualifiedTableName, tableAlias, backReferenceColumnInfos, qualifierColumnInfo,
-					idColumnInfos);
+					qualifierColumnType, idColumnInfos);
 
 		}
 
@@ -337,7 +362,7 @@ public interface AggregatePath extends Iterable<AggregatePath>, Comparable<Aggre
 			}
 		}
 
-		private static ColumnInfos computeReverseColumnInfo(AggregatePath path) {
+		private static ColumnInfos computeBackReferenceColumnInfos(AggregatePath path) {
 
 			AggregatePath tableOwner = AggregatePathTraversal.getTableOwningPath(path);
 
@@ -346,65 +371,79 @@ public interface AggregatePath extends Iterable<AggregatePath>, Comparable<Aggre
 			}
 
 			AggregatePath idDefiningParentPath = tableOwner.getIdDefiningParentPath();
-			RelationalPersistentProperty leafProperty = tableOwner.getRequiredLeafProperty();
+			RelationalPersistentProperty idProperty = idDefiningParentPath.getRequiredLeafEntity().getIdProperty();
-
-			RelationalPersistentProperty idProperty = idDefiningParentPath.getLeafEntity().getIdProperty();
 
-			if (idProperty != null) {
+			AggregatePath basePath = idProperty != null && idProperty.isEntity() ? idDefiningParentPath.append(idProperty)
-				if (idProperty.isEntity()) {
+					: idDefiningParentPath;
+			ColumInfosBuilder ciBuilder = new ColumInfosBuilder(basePath);
 
-					AggregatePath idBasePath = idDefiningParentPath.append(idProperty);
+			if (idProperty != null && idProperty.isEntity()) {
-					ColumInfosBuilder ciBuilder = new ColumInfosBuilder(idBasePath);
 
-					RelationalPersistentEntity<?> idEntity = idBasePath.getRequiredLeafEntity();
+				RelationalPersistentEntity<?> idEntity = basePath.getRequiredLeafEntity();
-					idEntity.doWithProperties((PropertyHandler<RelationalPersistentProperty>) p -> {
+				idEntity.doWithProperties((PropertyHandler<RelationalPersistentProperty>) p -> {
-						AggregatePath idElementPath = idBasePath.append(p);
+					AggregatePath idElementPath = basePath.append(p);
-						SqlIdentifier name = idElementPath.getColumnInfo().name();
+					SqlIdentifier name = idElementPath.getColumnInfo().name();
-						name = name.transform(n -> idDefiningParentPath.getTableInfo().qualifiedTableName.getReference() + "_" + n);
+					name = name.transform(n -> idDefiningParentPath.getTableInfo().qualifiedTableName.getReference() + "_" + n);
 
-						ciBuilder.add(idElementPath, name, name);
+					ciBuilder.add(idElementPath, name, name);
-					});
+				});
-
-					return ciBuilder.build();
-
-				} else {
-
-					ColumInfosBuilder ciBuilder = new ColumInfosBuilder(idDefiningParentPath);
-					SqlIdentifier reverseColumnName = leafProperty
-							.getReverseColumnName(idDefiningParentPath.getRequiredLeafEntity());
-
-					ciBuilder.add(idProperty, reverseColumnName,
-							AggregatePathTableUtils.prefixWithTableAlias(path, reverseColumnName));
 
-					return ciBuilder.build();
-				}
 			} else {
 
-				ColumInfosBuilder ciBuilder = new ColumInfosBuilder(idDefiningParentPath);
+				RelationalPersistentProperty leafProperty = tableOwner.getRequiredLeafProperty();
 				SqlIdentifier reverseColumnName = leafProperty
 						.getReverseColumnName(idDefiningParentPath.getRequiredLeafEntity());
+				SqlIdentifier alias = AggregatePathTableUtils.prefixWithTableAlias(path, reverseColumnName);
 
-				ciBuilder.add(idDefiningParentPath, reverseColumnName,
+				if (idProperty != null) {
-						AggregatePathTableUtils.prefixWithTableAlias(path, reverseColumnName));
+					ciBuilder.add(idProperty, reverseColumnName, alias);
-
+				} else {
-				return ciBuilder.build();
+					ciBuilder.add(idDefiningParentPath, reverseColumnName, alias);
+				}
 			}
+			return ciBuilder.build();
+		}
 
+		@Override
+		public ColumnInfos backReferenceColumnInfos() {
+			return backReferenceColumnInfos;
 		}
 
+		/**
+		 * Returns the unique {@link ColumnInfo} referencing the parent table, if such exists.
+		 *
+		 * @return guaranteed not to be {@literal null}.
+		 * @throws IllegalStateException if there is not exactly one back referencing column.
+		 * @deprecated since there might be more than one reverse column instead. Use {@link #backReferenceColumnInfos()}
+		 *             instead.
+		 */
 		@Deprecated(forRemoval = true)
 		public ColumnInfo reverseColumnInfo() {
-			return reverseColumnInfos.unique();
+			return backReferenceColumnInfos.unique();
 		}
 
+		/**
+		 * The id columns of the underlying table.
+		 * <p>
+		 * These might be:
+		 * <ul>
+		 * <li>the columns representing the id of the entity in question.</li>
+		 * <li>the columns representing the id of a parent entity, which _owns_ the table. Note that this case also covers
+		 * the first case.</li>
+		 * <li>or the backReferenceColumns.</li>
+		 * </ul>
+		 *
+		 * @return ColumnInfos representing the effective id of this entity. Guaranteed not to be {@literal null}.
+		 */
 		public ColumnInfos effectiveIdColumnInfos() {
-			return reverseColumnInfos.columnInfos.isEmpty() ? idColumnInfos : reverseColumnInfos;
+			return backReferenceColumnInfos.columnInfos.isEmpty() ? idColumnInfos : backReferenceColumnInfos;
 		}
 	}
 
 	/**
-	 * @param name The name of the column used to represent this property in the database.
+	 * @param name the name of the column used to represent this property in the database.
-	 * @param alias The alias for the column used to represent this property in the database.
+	 * @param alias the alias for the column used to represent this property in the database.
+	 * @since 3.2
 	 */
 	record ColumnInfo(SqlIdentifier name, SqlIdentifier alias) {
 
@@ -432,29 +471,49 @@ public interface AggregatePath extends Iterable<AggregatePath>, Comparable<Aggre
 	}
 
 	/**
-	 * A group of {@link ColumnInfo} values referenced by there respective {@link AggregatePath}. This is relevant for
+	 * A group of {@link ColumnInfo} values referenced by there respective {@link AggregatePath}. It is used in a similar
-	 * composite ids and references to such ids.
+	 * way as {@literal ColumnInfo} when one needs to consider more than a single column. This is relevant for composite
-	 **/
+	 * ids and references to such ids.
+	 *
+	 * @author Jens Schauder
+	 * @since 4.0
+	 */
 	class ColumnInfos {
 
 		private final AggregatePath basePath;
 		private final Map<AggregatePath, ColumnInfo> columnInfos;
+		private final Map<Table, List<Column>> columnCache = new HashMap<>();
 
 		/**
+		 * Creates a new ColumnInfos instances based on the arguments.
+		 *
 		 * @param basePath The path on which all other paths in the other argument are based on. For the typical case of a
 		 *          composite id, this would be the path to the composite ids.
 		 * @param columnInfos A map, mapping {@literal AggregatePath} instances to the respective {@literal ColumnInfo}
 		 */
-		private ColumnInfos(AggregatePath basePath, Map<AggregatePath, ColumnInfo> columnInfos) {
+		ColumnInfos(AggregatePath basePath, Map<AggregatePath, ColumnInfo> columnInfos) {
 
 			this.basePath = basePath;
 			this.columnInfos = columnInfos;
 		}
 
-		public static ColumnInfos empty(AggregatePath base) {
+		/**
-			return new ColumnInfos(base, new HashMap<>());
+		 * An empty {@literal ColumnInfos} instance with a fixed base path. Useful as a base when collecting
+		 * {@link ColumnInfo} instances into an {@literal ColumnInfos} instance.
+		 *
+		 * @param basePath The path on which paths in the {@literal ColumnInfos} or derived objects will be based on.
+		 * @return an empty instance save the {@literal basePath}.
+		 */
+		public static ColumnInfos empty(AggregatePath basePath) {
+			return new ColumnInfos(basePath, new HashMap<>());
 		}
 
+		/**
+		 * If this instance contains exactly one {@link ColumnInfo} it will be returned.
+		 *
+		 * @return the unique {@literal ColumnInfo} if present.
+		 * @throws IllegalStateException if the number of contained {@literal ColumnInfo} instances is not exactly 1.
+		 */
 		public ColumnInfo unique() {
 
 			Collection<ColumnInfo> values = columnInfos.values();
@@ -462,18 +521,41 @@ public interface AggregatePath extends Iterable<AggregatePath>, Comparable<Aggre
 			return values.iterator().next();
 		}
 
+		/**
+		 * Any of the contained {@link ColumnInfo} instances.
+		 *
+		 * @return a {@link ColumnInfo} instance.
+		 * @throws java.util.NoSuchElementException if no instance is available.
+		 */
 		public ColumnInfo any() {
 
 			Collection<ColumnInfo> values = columnInfos.values();
 			return values.iterator().next();
 		}
 
+		/**
+		 * Checks if {@literal this} instance is empty, i.e. does not contain any {@link ColumnInfo} instance.
+		 *
+		 * @return {@literal true} iff the collection of {@literal ColumnInfo} is empty.
+		 */
 		public boolean isEmpty() {
 			return columnInfos.isEmpty();
 		}
 
-		public <T> List<T> toList(Function<ColumnInfo, T> mapper) {
+		/**
-			return columnInfos.values().stream().map(mapper).toList();
+		 * Converts the given {@link Table} into a list of {@link Column}s. This method retrieves and caches the list of
+		 * columns for the specified table. If the columns are not already cached, it computes the list by mapping
+		 * {@code columnInfos} to their corresponding {@link Column} in the provided table and then stores the result in the
+		 * cache.
+		 *
+		 * @param table the {@link Table} for which the columns should be generated; must not be {@literal null}.
+		 * @return a list of {@link Column}s associated with the specified {@link Table}. Guaranteed no to be
+		 *         {@literal null}.
+		 */
+		public List<Column> toColumnList(Table table) {
+
+			return columnCache.computeIfAbsent(table,
+					t -> columnInfos.values().stream().map(columnInfo -> t.column(columnInfo.name)).toList());
 		}
 
 		/**
@@ -489,8 +571,8 @@ public interface AggregatePath extends Iterable<AggregatePath>, Comparable<Aggre
 		 *          an additional element into a result.
 		 * @param combiner an associative, non-interfering, stateless function for combining two values, which must be
 		 *          compatible with the {@code accumulator} function.
-		 * @return result of the function.
 		 * @param <T> type of the result.
+		 * @return result of the function.
 		 * @since 3.5
 		 */
 		public <T> T reduce(T identity, BiFunction<AggregatePath, ColumnInfo, T> accumulator, BinaryOperator<T> combiner) {
@@ -506,56 +588,57 @@ public interface AggregatePath extends Iterable<AggregatePath>, Comparable<Aggre
 			return result;
 		}
 
+		/**
+		 * Calls the consumer for each pair of {@link AggregatePath} and {@literal ColumnInfo}.
+		 *
+		 * @param consumer the function to call.
+		 */
 		public void forEach(BiConsumer<AggregatePath, ColumnInfo> consumer) {
 			columnInfos.forEach(consumer);
 		}
 
+		/**
+		 * Calls the {@literal mapper} for each pair one pair of {@link AggregatePath} and {@link ColumnInfo}, if there is
+		 * any.
+		 *
+		 * @param mapper the function to call.
+		 * @return the result of the mapper
+		 * @throws java.util.NoSuchElementException if this {@literal ColumnInfo} is empty.
+		 */
 		public <T> T any(BiFunction<AggregatePath, ColumnInfo, T> mapper) {
 
 			Map.Entry<AggregatePath, ColumnInfo> any = columnInfos.entrySet().iterator().next();
 			return mapper.apply(any.getKey(), any.getValue());
 		}
 
+		/**
+		 * Gets the {@link ColumnInfo} for the provided {@link AggregatePath}
+		 *
+		 * @param path for which to return the {@literal ColumnInfo}
+		 * @return {@literal ColumnInfo} for the given path.
+		 */
 		public ColumnInfo get(AggregatePath path) {
 			return columnInfos.get(path);
 		}
 
+		/**
+		 * Constructs an {@link AggregatePath} from the {@literal basePath} and the provided argument.
+		 *
+		 * @param ap {@literal AggregatePath} to be appended to the {@literal basePath}.
+		 * @return the combined (@literal AggregatePath}
+		 */
 		public AggregatePath fullPath(AggregatePath ap) {
 			return basePath.append(ap);
 		}
 
+		/**
+		 * Number of {@literal ColumnInfo} elements in this instance.
+		 *
+		 * @return the size of the collection of {@literal ColumnInfo}.
+		 */
 		public int size() {
 			return columnInfos.size();
 		}
 	}
 
-	class ColumInfosBuilder {
-		private final AggregatePath basePath;
-
-		private final Map<AggregatePath, ColumnInfo> columnInfoMap = new TreeMap<>();
-
-		public ColumInfosBuilder(AggregatePath basePath) {
-			this.basePath = basePath;
-		}
-
-		void add(AggregatePath path, SqlIdentifier name, SqlIdentifier alias) {
-			add(path, new ColumnInfo(name, alias));
-		}
-
-		public void add(RelationalPersistentProperty property, SqlIdentifier name, SqlIdentifier alias) {
-			add(basePath.append(property), name, alias);
-		}
-
-		ColumnInfos build() {
-			return new ColumnInfos(basePath, columnInfoMap);
-		}
-
-		public void add(AggregatePath path, ColumnInfo columnInfo) {
-			columnInfoMap.put(path.substract(basePath), columnInfo);
-		}
-	}
-
-	@Nullable
-	AggregatePath substract(@Nullable AggregatePath basePath);
-
 }

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

@@ -0,0 +1,84 @@
+/*
+ * Copyright 2025 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.Map;
+import java.util.TreeMap;
+
+import org.springframework.data.relational.core.sql.SqlIdentifier;
+
+/**
+ * A builder for {@link AggregatePath.ColumnInfos} instances.
+ *
+ * @author Jens Schauder
+ * @since 4.0
+ */
+class ColumInfosBuilder {
+
+	private final AggregatePath basePath;
+	private final Map<AggregatePath, AggregatePath.ColumnInfo> columnInfoMap = new TreeMap<>();
+
+	/**
+	 * Start construction with just the {@literal basePath} which all other paths are build upon.
+	 *
+	 * @param basePath must not be null.
+	 */
+	ColumInfosBuilder(AggregatePath basePath) {
+		this.basePath = basePath;
+	}
+
+	/**
+	 * Adds a {@link AggregatePath.ColumnInfo} to the {@link AggregatePath.ColumnInfos} under construction.
+	 *
+	 * @param path referencing the {@literal ColumnInfo}.
+	 * @param name of the column.
+	 * @param alias alias for the column.
+	 */
+	void add(AggregatePath path, SqlIdentifier name, SqlIdentifier alias) {
+		add(path, new AggregatePath.ColumnInfo(name, alias));
+	}
+
+	/**
+	 * Adds a {@link AggregatePath.ColumnInfo} to the {@link AggregatePath.ColumnInfos} under construction.
+	 *
+	 * @param property referencing the {@literal ColumnInfo}.
+	 * @param name of the column.
+	 * @param alias alias for the column.
+	 */
+	void add(RelationalPersistentProperty property, SqlIdentifier name, SqlIdentifier alias) {
+		add(basePath.append(property), name, alias);
+	}
+
+	/**
+	 * Adds a {@link AggregatePath.ColumnInfo} to the {@link AggregatePath.ColumnInfos} under construction.
+	 *
+	 * @param path the path referencing the {@literal ColumnInfo}
+	 * @param columnInfo the {@literal ColumnInfo} added.
+	 */
+	void add(AggregatePath path, AggregatePath.ColumnInfo columnInfo) {
+		columnInfoMap.put(path.subtract(basePath), columnInfo);
+	}
+
+	/**
+	 * Build the final {@link AggregatePath.ColumnInfos} instance.
+	 *
+	 * @return a {@literal ColumnInfos} instance containing all the added {@link AggregatePath.ColumnInfo} instances.
+	 */
+	AggregatePath.ColumnInfos build() {
+		return new AggregatePath.ColumnInfos(basePath, columnInfoMap);
+	}
+
+}

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

@@ -21,7 +21,6 @@ import java.util.Objects;
 
 import org.springframework.data.mapping.PersistentPropertyPath;
 import org.springframework.data.util.Lazy;
-import org.springframework.lang.NonNull;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.ConcurrentLruCache;
@@ -229,7 +228,7 @@ class DefaultAggregatePath implements AggregatePath {
 
 	@Override
 	@Nullable
-	public AggregatePath substract(@Nullable AggregatePath basePath) {
+	public AggregatePath subtract(@Nullable AggregatePath basePath) {
 
 		if (basePath == null || basePath.isRoot()) {
 			return this;
@@ -244,7 +243,7 @@ class DefaultAggregatePath implements AggregatePath {
 			if (tail == null) {
 				return null;
 			}
-			return tail.substract(basePath.getTail());
+			return tail.subtract(basePath.getTail());
 		}
 
 		throw new IllegalStateException("Can't subtract [%s] from [%s]".formatted(basePath, this));
@@ -303,11 +302,6 @@ class DefaultAggregatePath implements AggregatePath {
 				+ ((isRoot()) ? "/" : path.toDotPath());
 	}
 
-	@Override
-	public int compareTo(@NonNull AggregatePath other) {
-		return toDotPath().compareTo(other.toDotPath());
-	}
-
 	private static class AggregatePathIterator implements Iterator<AggregatePath> {
 
 		private @Nullable AggregatePath current;

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

@@ -50,7 +50,8 @@ public interface RelationalPersistentEntity<T> extends MutablePersistentEntity<T
 	 * Returns the column representing the identifier.
 	 *
 	 * @return will never be {@literal null}.
-	 * @deprecated use {@code AggregatePath.getTableInfo().getIdColumnInfos()} instead.
+	 * @deprecated because an entity may have multiple id columns. Use
+	 *             {@code AggregatePath.getTableInfo().getIdColumnInfos()} instead.
 	 */
 	@Deprecated(forRemoval = true)
 	SqlIdentifier getIdColumn();

spring-data-relational/src/main/java/org/springframework/data/relational/core/sql/AnalyticFunction.java

@@ -62,9 +62,9 @@ public class AnalyticFunction extends AbstractSegment implements Expression {
 	 * @param partitionBy Typically, column but other expressions are fine to.
 	 * @return a new {@literal AnalyticFunction} is partitioned by the given expressions, overwriting any expression
 	 *         previously present.
-	 * @since 3.5
+	 * @since 4.0
 	 */
-	public AnalyticFunction partitionBy(Collection<Expression> partitionBy) {
+	public AnalyticFunction partitionBy(Collection<? extends Expression> partitionBy) {
 		return partitionBy(partitionBy.toArray(new Expression[0]));
 	}
 
@@ -85,9 +85,9 @@ public class AnalyticFunction extends AbstractSegment implements Expression {
 	 * @param orderBy Typically, column but other expressions are fine to.
 	 * @return a new {@literal AnalyticFunction} is ordered by the given expressions, overwriting any expression
 	 *         previously present.
-	 * @since 3.5
+	 * @since 4.0
 	 */
-	public AnalyticFunction orderBy(Collection<Expression> orderBy) {
+	public AnalyticFunction orderBy(Collection<? extends Expression> orderBy) {
 		return orderBy(orderBy.toArray(new Expression[0]));
 	}
 

spring-data-relational/src/main/java/org/springframework/data/relational/core/sql/Expressions.java

@@ -15,15 +15,17 @@
  */
 package org.springframework.data.relational.core.sql;
 
+import java.util.List;
+
 /**
  * Factory for common {@link Expression}s.
  *
  * @author Mark Paluch
  * @author Jens Schauder
- * @since 1.1
  * @see SQL
  * @see Conditions
  * @see Functions
+ * @since 1.1
  */
 public abstract class Expressions {
 
@@ -61,6 +63,26 @@ public abstract class Expressions {
 		return Cast.create(expression, targetType);
 	}
 
+	/**
+	 * Creates an {@link Expression} based on the provided list of {@link Column}s.
+	 * <p>
+	 * If the list contains only a single column, this method returns that column directly as the resulting
+	 * {@link Expression}. Otherwise, it creates and returns a {@link TupleExpression} that represents multiple columns as
+	 * a single expression.
+	 *
+	 * @param columns the list of {@link Column}s to include in the expression; must not be {@literal null}.
+	 * @return an {@link Expression} corresponding to the input columns: either a single column or a
+	 *         {@link TupleExpression} for multiple columns.
+	 * @since 4.0
+	 */
+	public static Expression of(List<Column> columns) {
+
+		if (columns.size() == 1) {
+			return columns.get(0);
+		}
+		return new TupleExpression(columns);
+	}
+
 	// Utility constructor.
 	private Expressions() {}
 

spring-data-relational/src/main/java/org/springframework/data/relational/core/sql/TupleExpression.java

@@ -1,6 +1,20 @@
-package org.springframework.data.relational.core.sql;
+/*
+ * Copyright 2025 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.
+ */
 
-import org.jetbrains.annotations.NotNull;
+package org.springframework.data.relational.core.sql;
 
 import static java.util.stream.Collectors.*;
 
@@ -8,13 +22,13 @@ import java.util.List;
 
 /**
  * A tuple as used in conditions like
- * 
+ *
  * <pre>
  *   WHERE (one, two) IN (select x, y from some_table)
  * </pre>
  *
  * @author Jens Schauder
- * @since 3.5
+ * @since 4.0
  */
 public class TupleExpression extends AbstractSegment implements Expression {
 
@@ -24,7 +38,7 @@ public class TupleExpression extends AbstractSegment implements Expression {
 		return expressions.toArray(new Segment[0]);
 	}
 
-	private TupleExpression(List<? extends Expression> expressions) {
+	TupleExpression(List<? extends Expression> expressions) {
 
 		super(children(expressions));
 
@@ -39,14 +53,6 @@ public class TupleExpression extends AbstractSegment implements Expression {
 		return new TupleExpression(expressions);
 	}
 
-	public static Expression maybeWrap(List<Column> columns) {
-
-		if (columns.size() == 1) {
-			return columns.get(0);
-		}
-		return new TupleExpression(columns);
-	}
-
 	@Override
 	public String toString() {
 		return "(" + expressions.stream().map(Expression::toString).collect(joining(", ")) + ")";

spring-data-relational/src/main/java/org/springframework/data/relational/core/sql/render/TupleVisitor.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2024 the original author or authors.
+ * Copyright 2025 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.
@@ -22,7 +22,7 @@ import org.springframework.data.relational.core.sql.Visitable;
  * Visitor for rendering tuple expressions.
  *
  * @author Jens Schauder
- * @since 3.5
+ * @since 4.0
  */
 class TupleVisitor extends TypedSingleConditionRenderSupport<TupleExpression> implements PartRenderer {
 

spring-data-relational/src/main/java/org/springframework/data/relational/core/sqlgeneration/SingleQuerySqlGenerator.java

@@ -169,10 +169,7 @@ public class SingleQuerySqlGenerator implements SqlGenerator {
 		String rowCountAlias = aliases.getRowCountAlias(basePath);
 		Expression count = basePath.isRoot() ? new AliasedExpression(SQL.literalOf(1), rowCountAlias) //
 				: AnalyticFunction.create("count", Expressions.just("*")) //
-						.partitionBy( //
+						.partitionBy(basePath.getTableInfo().backReferenceColumnInfos().toColumnList(table) //
-								basePath.getTableInfo().reverseColumnInfos().toList( //
-										ci -> table.column(ci.name()) //
-								) //
 						).as(rowCountAlias);
 		columns.add(count);
 
@@ -182,7 +179,8 @@ public class SingleQuerySqlGenerator implements SqlGenerator {
 		if (!basePath.isRoot()) {
 
 			backReferenceAlias = aliases.getBackReferenceAlias(basePath);
-			columns.add(table.column(basePath.getTableInfo().reverseColumnInfos().unique().name()).as(backReferenceAlias));
+			columns
+					.add(table.column(basePath.getTableInfo().backReferenceColumnInfos().unique().name()).as(backReferenceAlias));
 
 			keyAlias = aliases.getKeyAlias(basePath);
 			Expression keyExpression = basePath.isQualified()
@@ -242,10 +240,10 @@ public class SingleQuerySqlGenerator implements SqlGenerator {
 
 	private static AnalyticFunction createRowNumberExpression(AggregatePath basePath, Table table,
 			String rowNumberAlias) {
-		AggregatePath.ColumnInfos reverseColumnInfos = basePath.getTableInfo().reverseColumnInfos();
+		AggregatePath.ColumnInfos reverseColumnInfos = basePath.getTableInfo().backReferenceColumnInfos();
 		return AnalyticFunction.create("row_number") //
-				.partitionBy(reverseColumnInfos.toList(ci -> table.column(ci.name()))) //
+				.partitionBy(reverseColumnInfos.toColumnList(table)) //
-				.orderBy(reverseColumnInfos.toList(ci -> table.column(ci.name()))) //
+				.orderBy(reverseColumnInfos.toColumnList(table)) //
 				.as(rowNumberAlias);
 	}
 

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

@@ -0,0 +1,80 @@
+/*
+ * Copyright 2025 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 org.assertj.core.api.AbstractAssert;
+
+/**
+ * Custom AssertJ assertions for {@link AggregatePath} instances
+ * 
+ * @author Jens Schauder
+ * @since 4.0
+ */
+public class AggregatePathAssertions extends AbstractAssert<AggregatePathAssertions, AggregatePath> {
+
+	/**
+	 * Constructor taking the actual {@link AggregatePath} to assert over.
+	 * 
+	 * @param actual
+	 */
+	public AggregatePathAssertions(AggregatePath actual) {
+		super(actual, AggregatePathAssertions.class);
+	}
+
+	/**
+	 * Entry point for creating assertions for AggregatePath.
+	 */
+	public static AggregatePathAssertions assertThat(AggregatePath actual) {
+		return new AggregatePathAssertions(actual);
+	}
+
+	/**
+	 * Assertion method comparing the path of the actual AggregatePath with the provided String representation of a path
+	 * in dot notation. Note that the assertion does not test the root entity type of the AggregatePath.
+	 */
+	public AggregatePathAssertions hasPath(String expectedPath) {
+		isNotNull();
+
+		if (!actual.toDotPath().equals(expectedPath)) { // Adjust this condition based on your AggregatePath's path logic
+			failWithMessage("Expected path to be <%s> but was <%s>", expectedPath, actual.toString());
+		}
+		return this;
+	}
+
+	/**
+	 * assertion testing if the actual path is a root path.
+	 */
+	public AggregatePathAssertions isRoot() {
+		isNotNull();
+
+		if (!actual.isRoot()) {
+			failWithMessage("Expected AggregatePath to be root path, but it was not");
+		}
+		return this;
+	}
+
+	/**
+	 * assertion testing if the actual path is NOT a root path.
+	 */
+	public AggregatePathAssertions isNotRoot() {
+		isNotNull();
+
+		if (actual.isRoot()) {
+			failWithMessage("Expected AggregatePath not to be root path, but it was.");
+		}
+		return this;
+	}
+}

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

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2025 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.Consumer;
+
+import org.assertj.core.api.SoftAssertions;
+import org.assertj.core.api.SoftAssertionsProvider;
+
+/**
+ * Soft assertions for {@link AggregatePath} instances.
+ *
+ * @author Jens Schauder
+ * @since 4.0
+ */
+public class AggregatePathSoftAssertions extends SoftAssertions {
+
+	/**
+	 * Entry point for assertions. The default {@literal assertThat} can't be used, since it collides with {@link SoftAssertions#assertThat(Iterable)}
+	 */
+	public AggregatePathAssertions assertAggregatePath(AggregatePath actual) {
+		return proxy(AggregatePathAssertions.class, AggregatePath.class, actual);
+	}
+
+	static void assertAggregatePathsSoftly(Consumer<AggregatePathSoftAssertions> softly) {
+		SoftAssertionsProvider.assertSoftly(AggregatePathSoftAssertions.class, softly);
+	}
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2024 the original author or authors.
+ * Copyright 2025 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.
@@ -26,14 +26,16 @@ import java.util.NoSuchElementException;
 import org.junit.jupiter.api.Test;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.relational.core.sql.SqlIdentifier;
+import org.springframework.data.relational.core.sql.Table;
 
 /**
  * Unit tests for the construction of {@link org.springframework.data.relational.core.mapping.AggregatePath.ColumnInfos}
- * 
+ *
  * @author Jens Schauder
  */
 class ColumnInfosUnitTests {
 
+	static final Table TABLE = Table.create("dummy");
 	static final SqlIdentifier ID = SqlIdentifier.quoted("ID");
 	RelationalMappingContext context = new RelationalMappingContext();
 
@@ -45,9 +47,7 @@ class ColumnInfosUnitTests {
 		assertThat(columnInfos.isEmpty()).isTrue();
 		assertThrows(NoSuchElementException.class, columnInfos::any);
 		assertThrows(IllegalStateException.class, columnInfos::unique);
-		assertThat(columnInfos.toList(ci -> {
+		assertThat(columnInfos.toColumnList(TABLE)).isEmpty();
-			throw new IllegalStateException("This should never get called");
-		})).isEmpty();
 	}
 
 	@Test // GH-574
@@ -58,21 +58,21 @@ class ColumnInfosUnitTests {
 		assertThat(columnInfos.isEmpty()).isFalse();
 		assertThat(columnInfos.any().name()).isEqualTo(ID);
 		assertThat(columnInfos.unique().name()).isEqualTo(ID);
-		assertThat(columnInfos.toList(ci -> ci.name())).containsExactly(ID);
+		assertThat(columnInfos.toColumnList(TABLE)).containsExactly(TABLE.column(ID));
 	}
 
 	@Test // GH-574
 	void multiElementColumnInfos() {
 
-		AggregatePath.ColumnInfos columnInfos = basePath(DummyEntityWithCompositeId.class).getTableInfo().idColumnInfos();
+		AggregatePath.ColumnInfos columnInfos = basePath(WithCompositeId.class).getTableInfo().idColumnInfos();
 
 		assertThat(columnInfos.isEmpty()).isFalse();
 		assertThat(columnInfos.any().name()).isEqualTo(SqlIdentifier.quoted("ONE"));
 		assertThrows(IllegalStateException.class, columnInfos::unique);
-		assertThat(columnInfos.toList(ci -> ci.name())) //
+		assertThat(columnInfos.toColumnList(TABLE)) //
 				.containsExactly( //
-						SqlIdentifier.quoted("ONE"), //
+						TABLE.column(SqlIdentifier.quoted("ONE")), //
-						SqlIdentifier.quoted("TWO") //
+						TABLE.column(SqlIdentifier.quoted("TWO")) //
 				);
 
 		List<String> collector = new ArrayList<>();
@@ -97,6 +97,6 @@ class ColumnInfosUnitTests {
 	record CompositeId(String one, String two) {
 	}
 
-	record DummyEntityWithCompositeId(@Id @Embedded.Nullable CompositeId id, String name) {
+	record WithCompositeId(@Id @Embedded.Nullable CompositeId id, String name) {
 	}
 }

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

@@ -30,7 +30,9 @@ 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.Column;
 import org.springframework.data.relational.core.sql.SqlIdentifier;
+import org.springframework.data.relational.core.sql.Table;
 
 /**
  * Tests for {@link AggregatePath}.
@@ -48,7 +50,7 @@ class DefaultAggregatePathUnitTests {
 
 		AggregatePath path = context.getAggregatePath(context.getPersistentPropertyPath("entityId", DummyEntity.class));
 
-		assertThat(path.isRoot()).isFalse();
+		AggregatePathAssertions.assertThat(path).isNotRoot();
 	}
 
 	@Test // GH-1525
@@ -56,17 +58,17 @@ class DefaultAggregatePathUnitTests {
 
 		AggregatePath path = context.getAggregatePath(entity);
 
-		assertThat(path.isRoot()).isTrue();
+		AggregatePathAssertions.assertThat(path).isRoot();
 	}
 
 	@Test // GH-1525
 	void getParentPath() {
 
-		assertSoftly(softly -> {
+		AggregatePathSoftAssertions.assertAggregatePathsSoftly(softly -> {
 
-			softly.assertThat((Object) path("second.third2.value").getParentPath()).isEqualTo(path("second.third2"));
+			softly.assertAggregatePath(path("second.third2.value").getParentPath()).hasPath("second.third2");
-			softly.assertThat((Object) path("second.third2").getParentPath()).isEqualTo(path("second"));
+			softly.assertAggregatePath(path("second.third2").getParentPath()).hasPath("second");
-			softly.assertThat((Object) path("second").getParentPath()).isEqualTo(path());
+			softly.assertAggregatePath(path("second").getParentPath()).isRoot();
 
 			softly.assertThatThrownBy(() -> path().getParentPath()).isInstanceOf(IllegalStateException.class);
 		});
@@ -77,13 +79,13 @@ class DefaultAggregatePathUnitTests {
 
 		assertSoftly(softly -> {
 
+			RelationalPersistentEntity<?> secondEntity = context.getRequiredPersistentEntity(Second.class);
+			RelationalPersistentEntity<?> thirdEntity = context.getRequiredPersistentEntity(Third.class);
+
 			softly.assertThat(path().getRequiredLeafEntity()).isEqualTo(entity);
-			softly.assertThat(path("second").getRequiredLeafEntity())
+			softly.assertThat(path("second").getRequiredLeafEntity()).isEqualTo(secondEntity);
-					.isEqualTo(context.getRequiredPersistentEntity(Second.class));
+			softly.assertThat(path("second.third").getRequiredLeafEntity()).isEqualTo(thirdEntity);
-			softly.assertThat(path("second.third").getRequiredLeafEntity())
+			softly.assertThat(path("secondList").getRequiredLeafEntity()).isEqualTo(secondEntity);
-					.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);
@@ -94,17 +96,16 @@ class DefaultAggregatePathUnitTests {
 	@Test // GH-1525
 	void idDefiningPath() {
 
-		assertSoftly(softly -> {
+		AggregatePathSoftAssertions.assertAggregatePathsSoftly(softly -> {
 
-			softly.assertThat((Object) path("second.third2.value").getIdDefiningParentPath()).isEqualTo(path());
+			softly.assertAggregatePath(path("second.third2.value").getIdDefiningParentPath()).isRoot();
-			softly.assertThat((Object) path("second.third.value").getIdDefiningParentPath()).isEqualTo(path());
+			softly.assertAggregatePath(path("second.third.value").getIdDefiningParentPath()).isRoot();
-			softly.assertThat((Object) path("secondList.third2.value").getIdDefiningParentPath()).isEqualTo(path());
+			softly.assertAggregatePath(path("secondList.third2.value").getIdDefiningParentPath()).isRoot();
-			softly.assertThat((Object) path("secondList.third.value").getIdDefiningParentPath()).isEqualTo(path());
+			softly.assertAggregatePath(path("secondList.third.value").getIdDefiningParentPath()).isRoot();
-			softly.assertThat((Object) path("second2.third2.value").getIdDefiningParentPath()).isEqualTo(path());
+			softly.assertAggregatePath(path("second2.third2.value").getIdDefiningParentPath()).isRoot();
-			softly.assertThat((Object) path("second2.third.value").getIdDefiningParentPath()).isEqualTo(path());
+			softly.assertAggregatePath(path("second2.third.value").getIdDefiningParentPath()).isRoot();
-			softly.assertThat((Object) path("withId.second.third2.value").getIdDefiningParentPath())
+			softly.assertAggregatePath(path("withId.second.third2.value").getIdDefiningParentPath()).hasPath("withId");
-					.isEqualTo(path("withId"));
+			softly.assertAggregatePath(path("withId.second.third.value").getIdDefiningParentPath()).hasPath("withId");
-			softly.assertThat((Object) path("withId.second.third.value").getIdDefiningParentPath()).isEqualTo(path("withId"));
 		});
 	}
 
@@ -147,8 +148,10 @@ class DefaultAggregatePathUnitTests {
 	void reverseColumnNames() {
 
 		assertSoftly(softly -> {
-			softly.assertThat(path(CompoundIdEntity.class, "second").getTableInfo().reverseColumnInfos().toList(x -> x))
+			softly
-					.extracting(AggregatePath.ColumnInfo::name)
+					.assertThat(path(CompoundIdEntity.class, "second").getTableInfo().backReferenceColumnInfos()
+							.toColumnList(Table.create("dummy")))
+					.extracting(Column::getName)
 					.containsExactlyInAnyOrder(quoted("COMPOUND_ID_ENTITY_ONE"), quoted("COMPOUND_ID_ENTITY_TWO"));
 
 		});
@@ -183,13 +186,11 @@ class DefaultAggregatePathUnitTests {
 
 	@Test // GH-1525
 	void extendBy() {
+		AggregatePathSoftAssertions.assertAggregatePathsSoftly(softly -> {
 
-		assertSoftly(softly -> {
+			softly.assertAggregatePath(path().append(entity.getRequiredPersistentProperty("withId"))).hasPath("withId");
-
+			softly.assertAggregatePath(path("withId").append(path("withId").getRequiredIdProperty()))
-			softly.assertThat((Object) path().append(entity.getRequiredPersistentProperty("withId")))
+					.hasPath("withId.withIdId");
-					.isEqualTo(path("withId"));
-			softly.assertThat((Object) path("withId").append(path("withId").getRequiredIdProperty()))
-					.isEqualTo(path("withId.withIdId"));
 		});
 	}
 
@@ -244,11 +245,11 @@ class DefaultAggregatePathUnitTests {
 			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()).isTrue();
-			// softly.assertThat(path("secondList.third.value").isMultiValued()).isFalse();
 			softly.assertThat(path("secondList").isMultiValued()).isTrue();
 		});
 	}
@@ -453,8 +454,7 @@ class DefaultAggregatePathUnitTests {
 		});
 	}
 
-	@Test
+	@Test // GH-1525
-	// GH-1525
 	void getLength() {
 
 		assertSoftly(softly -> {
@@ -472,25 +472,24 @@ class DefaultAggregatePathUnitTests {
 	@Test // GH-574
 	void getTail() {
 
-		assertSoftly(softly -> {
+		AggregatePathSoftAssertions.assertAggregatePathsSoftly(softly -> {
 
-			softly.assertThat((Object) path().getTail()).isEqualTo(null);
+			softly.assertAggregatePath(path().getTail()).isNull();
-			softly.assertThat((Object) path("second").getTail()).isEqualTo(null);
+			softly.assertAggregatePath(path("second").getTail()).isNull();
-			softly.assertThat(path("second.third").getTail().toDotPath()).isEqualTo("third");
+			softly.assertAggregatePath(path("second.third").getTail()).hasPath("third");
-			softly.assertThat(path("second.third.value").getTail().toDotPath()).isEqualTo("third.value");
+			softly.assertAggregatePath(path("second.third.value").getTail()).hasPath("third.value");
 		});
 	}
 
 	@Test // GH-74
 	void append() {
 
-		assertSoftly(softly -> {
+		AggregatePathSoftAssertions.assertAggregatePathsSoftly(softly -> {
-
+			softly.assertAggregatePath(path("second").append(path())).hasPath("second");
-			softly.assertThat(path("second").append(path()).toDotPath()).isEqualTo("second");
+			softly.assertAggregatePath(path().append(path("second"))).hasPath("second");
-			softly.assertThat(path().append(path("second")).toDotPath()).isEqualTo("second");
+			softly.assertAggregatePath(path().append(path("second.third"))).hasPath("second.third");
-			softly.assertThat(path().append(path("second.third")).toDotPath()).isEqualTo("second.third");
 			AggregatePath value = path("second.third.value").getTail().getTail();
-			softly.assertThat(path("second.third").append(value).toDotPath()).isEqualTo("second.third.value");
+			softly.assertAggregatePath(path("second.third").append(value)).hasPath("second.third.value");
 		});
 	}
 

spring-data-relational/src/test/java/org/springframework/data/relational/core/sql/TupleExpressionUnitTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2024 the original author or authors.
+ * Copyright 2025 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.
@@ -22,6 +22,11 @@ import java.util.List;
 
 import org.junit.jupiter.api.Test;
 
+/**
+ * Unit tests for construction of {@link TupleExpression}.
+ *
+ * @author Jens Schauder
+ */
 class TupleExpressionUnitTests {
 
 	@Test // GH-574
@@ -29,7 +34,7 @@ class TupleExpressionUnitTests {
 
 		Column testColumn = Column.create("name", Table.create("employee"));
 
-		Expression wrapped = TupleExpression.maybeWrap(List.of(testColumn));
+		Expression wrapped = Expressions.of(List.of(testColumn));
 
 		assertThat(wrapped).isSameAs(testColumn);
 	}
@@ -40,7 +45,7 @@ class TupleExpressionUnitTests {
 		Column testColumn1 = Column.create("first", Table.create("employee"));
 		Column testColumn2 = Column.create("last", Table.create("employee"));
 
-		Expression wrapped = TupleExpression.maybeWrap(List.of(testColumn1, testColumn2));
+		Expression wrapped = Expressions.of(List.of(testColumn1, testColumn2));
 
 		assertThat(wrapped).isInstanceOf(TupleExpression.class);
 	}

src/main/antora/modules/ROOT/pages/jdbc/mapping.adoc

@@ -65,7 +65,8 @@ The table of the referenced entity is expected to have an additional column with
 * `Map<simple type, some entity>` is considered a qualified one-to-many relationship.
 The table of the referenced entity is expected to have two additional columns: One named based on the referencing entity for the foreign key (see <<jdbc.entity-persistence.types.backrefs>>) and one with the same name and an additional `_key` suffix for the map key.
 
-* `List<some entity>` is mapped as a  `Map<Integer, some entity>`. The same additional columns are expected and the names used can be customized in the same way.
+* `List<some entity>` is mapped as a  `Map<Integer, some entity>`.
+The same additional columns are expected and the names used can be customized in the same way.
 +
 For `List`, `Set`, and `Map` naming of the back reference can be controlled by implementing `NamingStrategy.getReverseColumnName(RelationalPersistentEntity<?> owner)` and `NamingStrategy.getKeyColumn(RelationalPersistentProperty property)`, respectively.
 Alternatively you may annotate the attribute with `@MappedCollection(idColumn="your_column_name", keyColumn="your_key_column_name")`.

src/main/antora/modules/ROOT/partials/mapping-annotations.adoc

@@ -1,7 +1,17 @@
 The `RelationalConverter` can use metadata to drive the mapping of objects to rows.
 The following annotations are available:
 
+* `@Embedded`: an entity with this annotation will be mapped to the table of the parent entity, instead of a separate table.
+Allows to specify if the resulting columns should have a common prefix.
+If all columns resulting from such an entity are `null` either the annotated entity will be `null` or _empty_, i.e. all of its properties will be `null`, depending on the value of `@Embedded.onEmpty()`
+May be combined with `@Id` to form a composite id.
 * `@Id`: Applied at the field level to mark the primary key.
+It may be combined with `@Embedded` to form a composite id.
+* `@InsertOnlyProperty`: Marks a property as only to be written during insert.
+Such a property on an aggregate root will only be written once and never updated.
+Note that on a nested entity, all save operations result in an insert therefore this annotation has no effect on properties of nested entities.
+* `@MappedCollection`: Allows for configuration how a collection, or a single nested entity gets mapped. `idColumn` specifies the column used for referencing the parent entities primary key. `keyColumn` specifies the column used to store the index of a `List` or the key of a `Map`.
+* `@Sequence`: specify a database sequence for generating values for the annotated property.
 * `@Table`: Applied at the class level to indicate this class is a candidate for mapping to the database.
 You can specify the name of the table where the database is stored.
 * `@Transient`: By default, all fields are mapped to the row.
@@ -22,3 +32,5 @@ However, this is not recommended, since it may cause problems with other tools.
 The value is `null` (`zero` for primitive types) is considered as marker for entities to be new.
 The initially stored value is `zero` (`one` for primitive types).
 The version gets incremented automatically on every update.
+
+

src/main/antora/modules/ROOT/partials/mapping.adoc

@@ -88,7 +88,6 @@ endif::[]
 You may use xref:value-expressions.adoc[Spring Data's SpEL support] to dynamically create column names.
 Once generated the names will be cached, so it is dynamic per mapping context only.
 
-
 ifdef::embedded-entities[]
 
 [[entity-persistence.embedded-entities]]
@@ -156,6 +155,43 @@ Entities may be annotated with `@Id` and `@Embedded`, resulting in a composite i
 The full embedded entity is considered the id, and therefore the check for determining if an aggregate is considered a new aggregate requiring an insert or an existing one, asking for an update is based on that entity, not its elements.
 Most use cases will require a custom `BeforeConvertCallback` to set the id for new aggregate.
 
+====
+.Simple entity with composite id
+[source,java]
+----
+@Table("PERSON_WITH_COMPOSITE_ID")
+record Person( <1>
+    @Id @Embedded.Nullable Name pk, <2>
+    String nickName,
+    Integer age
+) {
+}
+
+record Name(String first, String last) {
+}
+----
+
+.Matching table for simple entity with composite id
+[source,sql]
+----
+CREATE TABLE PERSON_WITH_COMPOSITE_ID (
+    FIRST VARCHAR(100),
+    LAST VARCHAR(100),
+    NICK_NAME VARCHAR(100),
+    AGE INT,
+    PRIMARY KEY (FIRST, LAST) <3>
+);
+
+
+----
+
+<1> Entities may be represented as records without any special consideration
+<2> `pk` is marked as id and embedded
+<3> the two columns from the embedded `Name` entity make up the primary key in the database.
+
+Details of the create tables will depend on the database used.
+====
+
 [[entity-persistence.read-only-properties]]
 == Read Only Properties