Add support for composite ids.

Entities may be annotated with `@Id` and `@Embedded`, resulting in a composite id on the database side.
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.

For an entity with `@Embedded` id, the back reference used in tables for referenced entities consists of multiple columns, each named by a concatenation of <table-name> + `_` + <column-name>.
E.g. the back reference to a `Person` entity, with a composite id with the properties `firstName` and `lastName` will consist of the two columns `PERSON_FIRST_NAME` and `PERSON_LAST_NAME`.
This holds for directly referenced entities as well as `List`, `Set` and `Map`.

Closes #574

Author: schauder

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

@@ -115,6 +115,7 @@ public <T> Object insert(T instance, Class<T> domainType, Identifier identifier,
 	public <T> Object[] insert(List<InsertSubject<T>> insertSubjects, Class<T> domainType, IdValueSource idValueSource) {
 
 		Assert.notEmpty(insertSubjects, "Batch insert must contain at least one InsertSubject");
+
 		SqlIdentifierParameterSource[] sqlParameterSources = insertSubjects.stream()
 				.map(insertSubject -> sqlParametersFactory.forInsert(insertSubject.getInstance(), domainType,
 						insertSubject.getIdentifier(), idValueSource))
@@ -160,7 +161,7 @@ public <S> boolean updateWithVersion(S instance, Class<S> domainType, Number pre
 	public void delete(Object id, Class<?> domainType) {
 
 		String deleteByIdSql = sql(domainType).getDeleteById();
-		SqlParameterSource parameter = sqlParametersFactory.forQueryById(id, domainType, ID_SQL_PARAMETER);
+		SqlParameterSource parameter = sqlParametersFactory.forQueryById(id, domainType);
 
 		operations.update(deleteByIdSql, parameter);
 	}
@@ -181,7 +182,7 @@ public <T> void deleteWithVersion(Object id, Class<T> domainType, Number previou
 
 		RelationalPersistentEntity<T> persistentEntity = getRequiredPersistentEntity(domainType);
 
-		SqlIdentifierParameterSource parameterSource = sqlParametersFactory.forQueryById(id, domainType, ID_SQL_PARAMETER);
+		SqlIdentifierParameterSource parameterSource = sqlParametersFactory.forQueryById(id, domainType);
 		parameterSource.addValue(VERSION_SQL_PARAMETER, previousVersion);
 		int affectedRows = operations.update(sql(domainType).getDeleteByIdAndVersion(), parameterSource);
 
@@ -201,8 +202,7 @@ public void delete(Object rootId, PersistentPropertyPath<RelationalPersistentPro
 
 		String delete = sql(rootEntity.getType()).createDeleteByPath(propertyPath);
 
-		SqlIdentifierParameterSource parameters = sqlParametersFactory.forQueryById(rootId, rootEntity.getType(),
-				ROOT_ID_PARAMETER);
+		SqlIdentifierParameterSource parameters = sqlParametersFactory.forQueryById(rootId, rootEntity.getType());
 		operations.update(delete, parameters);
 	}
 
@@ -236,7 +236,7 @@ public void deleteAll(PersistentPropertyPath<RelationalPersistentProperty> prope
 	public <T> void acquireLockById(Object id, LockMode lockMode, Class<T> domainType) {
 
 		String acquireLockByIdSql = sql(domainType).getAcquireLockById(lockMode);
-		SqlIdentifierParameterSource parameter = sqlParametersFactory.forQueryById(id, domainType, ID_SQL_PARAMETER);
+		SqlIdentifierParameterSource parameter = sqlParametersFactory.forQueryById(id, domainType);
 
 		operations.query(acquireLockByIdSql, parameter, ResultSet::next);
 	}
@@ -262,7 +262,7 @@ public long count(Class<?> domainType) {
 	public <T> T findById(Object id, Class<T> domainType) {
 
 		String findOneSql = sql(domainType).getFindOne();
-		SqlIdentifierParameterSource parameter = sqlParametersFactory.forQueryById(id, domainType, ID_SQL_PARAMETER);
+		SqlIdentifierParameterSource parameter = sqlParametersFactory.forQueryById(id, domainType);
 
 		try {
 			return operations.queryForObject(findOneSql, parameter, getEntityRowMapper(domainType));
@@ -329,7 +329,7 @@ public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
 	public <T> boolean existsById(Object id, Class<T> domainType) {
 
 		String existsSql = sql(domainType).getExists();
-		SqlParameterSource parameter = sqlParametersFactory.forQueryById(id, domainType, ID_SQL_PARAMETER);
+		SqlParameterSource parameter = sqlParametersFactory.forQueryById(id, domainType);
 
 		Boolean result = operations.queryForObject(existsSql, parameter, Boolean.class);
 		Assert.state(result != null, "The result of an exists query must not be null");

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

@@ -15,7 +15,13 @@
  */
 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;
 
 /**
@@ -41,13 +47,31 @@ public static JdbcIdentifierBuilder empty() {
 	 */
 	public static JdbcIdentifierBuilder forBackReferences(JdbcConverter converter, AggregatePath path, Object value) {
 
-		Identifier identifier = Identifier.of( //
-				path.getTableInfo().reverseColumnInfo().name(), //
-				value, //
-				converter.getColumnType(path.getIdDefiningParentPath().getRequiredIdProperty()) //
-		);
+		RelationalPersistentProperty idProperty = path.getIdDefiningParentPath().getRequiredIdProperty();
+		AggregatePath.ColumnInfos reverseColumnInfos = path.getTableInfo().reverseColumnInfos();
+
+		// create property accessor
+		RelationalMappingContext mappingContext = converter.getMappingContext();
+		RelationalPersistentEntity<?> persistentEntity = mappingContext.getPersistentEntity(idProperty.getType());
+
+		Function<AggregatePath, Object> valueProvider;
+		if (persistentEntity == null) {
+			valueProvider = ap -> value;
+		} else {
+			PersistentPropertyPathAccessor<Object> propertyPathAccessor = persistentEntity.getPropertyPathAccessor(value);
+			valueProvider = ap -> propertyPathAccessor.getProperty(ap.getRequiredPersistentPropertyPath());
+		}
+
+		final Identifier[] identifierHolder = new Identifier[] { Identifier.empty() };
+
+		reverseColumnInfos.forEach((ap, ci) -> {
+
+			RelationalPersistentProperty property = ap.getRequiredLeafProperty();
+			identifierHolder[0] = identifierHolder[0].withPart(ci.name(), valueProvider.apply(ap),
+					converter.getColumnType(property));
+		});
 
-		return new JdbcIdentifierBuilder(identifier);
+		return new JdbcIdentifierBuilder(identifierHolder[0]);
 	}
 
 	/**
@@ -62,8 +86,9 @@ public JdbcIdentifierBuilder withQualifier(AggregatePath path, Object value) {
 		Assert.notNull(path, "Path must not be null");
 		Assert.notNull(value, "Value must not be null");
 
-		identifier = identifier.withPart(path.getTableInfo().qualifierColumnInfo().name(), value,
-				path.getTableInfo().qualifierColumnType());
+		AggregatePath.TableInfo tableInfo = path.getTableInfo();
+		identifier = identifier.withPart(tableInfo.qualifierColumnInfo().name(), value,
+				tableInfo.qualifierColumnType());
 
 		return this;
 	}

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

@@ -80,7 +80,7 @@ public class MappingJdbcConverter extends MappingRelationalConverter implements
 	 * {@link #MappingJdbcConverter(RelationalMappingContext, RelationResolver, CustomConversions, JdbcTypeFactory)}
 	 * (MappingContext, RelationResolver, JdbcTypeFactory)} to convert arrays and large objects into JDBC-specific types.
 	 *
-	 * @param context          must not be {@literal null}.
+	 * @param context must not be {@literal null}.
 	 * @param relationResolver used to fetch additional relations from the database. Must not be {@literal null}.
 	 */
 	public MappingJdbcConverter(RelationalMappingContext context, RelationResolver relationResolver) {
@@ -98,12 +98,12 @@ public MappingJdbcConverter(RelationalMappingContext context, RelationResolver r
 	/**
 	 * Creates a new {@link MappingJdbcConverter} given {@link MappingContext}.
 	 *
-	 * @param context          must not be {@literal null}.
+	 * @param context must not be {@literal null}.
 	 * @param relationResolver used to fetch additional relations from the database. Must not be {@literal null}.
-	 * @param typeFactory      must not be {@literal null}
+	 * @param typeFactory must not be {@literal null}
 	 */
 	public MappingJdbcConverter(RelationalMappingContext context, RelationResolver relationResolver,
-								CustomConversions conversions, JdbcTypeFactory typeFactory) {
+			CustomConversions conversions, JdbcTypeFactory typeFactory) {
 
 		super(context, conversions);
 
@@ -220,7 +220,7 @@ private boolean canWriteAsJdbcValue(@Nullable Object value) {
 			return true;
 		}
 
-		if (value instanceof AggregateReference aggregateReference) {
+		if (value instanceof AggregateReference<?, ?> aggregateReference) {
 			return canWriteAsJdbcValue(aggregateReference.getId());
 		}
 
@@ -285,7 +285,7 @@ public <R> R readAndResolve(TypeInformation<R> type, RowDocument source, Identif
 
 	@Override
 	protected RelationalPropertyValueProvider newValueProvider(RowDocumentAccessor documentAccessor,
-															   ValueExpressionEvaluator evaluator, ConversionContext context) {
+			ValueExpressionEvaluator evaluator, ConversionContext context) {
 
 		if (context instanceof ResolvingConversionContext rcc) {
 
@@ -314,7 +314,7 @@ class ResolvingRelationalPropertyValueProvider implements RelationalPropertyValu
 		private final Identifier identifier;
 
 		private ResolvingRelationalPropertyValueProvider(AggregatePathValueProvider delegate, RowDocumentAccessor accessor,
-														 ResolvingConversionContext context, Identifier identifier) {
+				ResolvingConversionContext context, Identifier identifier) {
 
 			AggregatePath path = context.aggregatePath();
 
@@ -323,15 +323,15 @@ private ResolvingRelationalPropertyValueProvider(AggregatePathValueProvider dele
 			this.context = context;
 			this.identifier = path.isEntity()
 					? potentiallyAppendIdentifier(identifier, path.getRequiredLeafEntity(),
-					property -> delegate.getValue(path.append(property)))
+							property -> delegate.getValue(path.append(property)))
 					: identifier;
 		}
 
 		/**
 		 * Conditionally append the identifier if the entity has an identifier property.
 		 */
 		static Identifier potentiallyAppendIdentifier(Identifier base, RelationalPersistentEntity<?> entity,
-													  Function<RelationalPersistentProperty, Object> getter) {
+				Function<RelationalPersistentProperty, Object> getter) {
 
 			if (entity.hasIdProperty()) {
 
@@ -361,24 +361,9 @@ public <T> T getPropertyValue(RelationalPersistentProperty property) {
 
 				if (property.isCollectionLike() || property.isMap()) {
 
-					Identifier identifierToUse = this.identifier;
-					AggregatePath idDefiningParentPath = aggregatePath.getIdDefiningParentPath();
+					Identifier identifier = constructIdentifier(aggregatePath);
 
-					// 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 identifier = idDefiningParentPath.getRequiredIdProperty();
-						AggregatePath idPath = idDefiningParentPath.append(identifier);
-						Object value = delegate.getValue(idPath);
-
-						Assert.state(value != null, "Identifier value must not be null at this point");
-
-						identifierToUse = Identifier.of(aggregatePath.getTableInfo().reverseColumnInfo().name(), value,
-								identifier.getActualType());
-					}
-
-					Iterable<Object> allByPath = relationResolver.findAllByPath(identifierToUse,
+					Iterable<Object> allByPath = relationResolver.findAllByPath(identifier,
 							aggregatePath.getRequiredPersistentPropertyPath());
 
 					if (property.isCollectionLike()) {
@@ -403,6 +388,29 @@ public <T> T getPropertyValue(RelationalPersistentProperty property) {
 			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) {
 
@@ -423,7 +431,7 @@ public boolean hasValue(RelationalPersistentProperty property) {
 					return delegate.hasValue(toUse);
 				}
 
-				return delegate.hasValue(aggregatePath.getTableInfo().reverseColumnInfo().alias());
+				return delegate.hasValue(aggregatePath.getTableInfo().reverseColumnInfos().any().alias());
 			}
 
 			return delegate.hasValue(aggregatePath);
@@ -449,7 +457,7 @@ public boolean hasNonEmptyValue(RelationalPersistentProperty property) {
 					return delegate.hasValue(toUse);
 				}
 
-				return delegate.hasValue(aggregatePath.getTableInfo().reverseColumnInfo().alias());
+				return delegate.hasValue(aggregatePath.getTableInfo().reverseColumnInfos().any().alias());
 			}
 
 			return delegate.hasNonEmptyValue(aggregatePath);
@@ -460,7 +468,7 @@ public RelationalPropertyValueProvider withContext(ConversionContext context) {
 
 			return context == this.context ? this
 					: new ResolvingRelationalPropertyValueProvider(delegate.withContext(context), accessor,
-					(ResolvingConversionContext) context, identifier);
+							(ResolvingConversionContext) context, identifier);
 		}
 	}
 
@@ -472,7 +480,7 @@ public RelationalPropertyValueProvider withContext(ConversionContext context) {
 	 * @param identifier
 	 */
 	private record ResolvingConversionContext(ConversionContext delegate, AggregatePath aggregatePath,
-											  Identifier identifier) implements ConversionContext {
+			Identifier identifier) implements ConversionContext {
 
 		@Override
 		public <S> S convert(Object source, TypeInformation<? extends S> typeHint) {

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

@@ -40,10 +40,6 @@ class SqlContext {
 		this.table = Table.create(entity.getQualifiedTableName());
 	}
 
-	Column getIdColumn() {
-		return table.column(entity.getIdColumn());
-	}
-
 	Column getVersionColumn() {
 		return table.column(entity.getRequiredVersionProperty().getColumnName());
 	}
@@ -60,11 +56,20 @@ Table getTable(AggregatePath path) {
 	}
 
 	Column getColumn(AggregatePath path) {
+
 		AggregatePath.ColumnInfo columnInfo = path.getColumnInfo();
 		return getTable(path).column(columnInfo.name()).as(columnInfo.alias());
 	}
 
-	Column getReverseColumn(AggregatePath path) {
-		return getTable(path).column(path.getTableInfo().reverseColumnInfo().name()).as(path.getTableInfo().reverseColumnInfo().alias());
+	/**
+	 * A token reverse column, used in selects to identify, if an entity is present or {@literal null}.
+	 * 
+	 * @param path must not be null.
+	 * @return a {@literal Column} that is part of the effective primary key for the given path.
+	 */
+	Column getAnyReverseColumn(AggregatePath path) {
+
+		AggregatePath.ColumnInfo columnInfo = path.getTableInfo().reverseColumnInfos().any();
+		return getTable(path).column(columnInfo.name()).as(columnInfo.alias());
 	}
 }