Add support for composite ids

pom.xml

@@ -5,7 +5,7 @@
 
 	<groupId>org.springframework.data</groupId>
 	<artifactId>spring-data-relational-parent</artifactId>
-	<version>3.5.0-SNAPSHOT</version>
+	<version>3.5.0-1737-nullable-embedded-with-collection-574-composite-id-SNAPSHOT</version>
 	<packaging>pom</packaging>
 
 	<name>Spring Data Relational Parent</name>

spring-data-jdbc-distribution/pom.xml

@@ -14,7 +14,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-relational-parent</artifactId>
-		<version>3.5.0-SNAPSHOT</version>
+		<version>3.5.0-1737-nullable-embedded-with-collection-574-composite-id-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

spring-data-jdbc/pom.xml

@@ -6,7 +6,7 @@
 	<modelVersion>4.0.0</modelVersion>
 
 	<artifactId>spring-data-jdbc</artifactId>
-	<version>3.5.0-SNAPSHOT</version>
+	<version>3.5.0-1737-nullable-embedded-with-collection-574-composite-id-SNAPSHOT</version>
 
 	<name>Spring Data JDBC</name>
 	<description>Spring Data module for JDBC repositories.</description>
@@ -15,7 +15,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-relational-parent</artifactId>
-		<version>3.5.0-SNAPSHOT</version>
+		<version>3.5.0-1737-nullable-embedded-with-collection-574-composite-id-SNAPSHOT</version>
 	</parent>
 
 	<properties>

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

@@ -117,6 +117,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(), //
@@ -167,7 +168,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);
 	}
@@ -188,7 +189,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);
 
@@ -208,8 +209,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);
 	}
 
@@ -243,7 +243,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);
 	}
@@ -269,7 +269,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));
@@ -355,7 +355,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/Identifier.java

@@ -99,6 +99,25 @@ public static Identifier from(Map<SqlIdentifier, Object> map) {
 		return new Identifier(Collections.unmodifiableList(values));
 	}
 
+	/**
+	 * Creates a new {@link Identifier} from the current instance and sets the value from {@link Identifier}. Existing key
+	 * definitions for {@code name} are overwritten if they already exist.
+	 *
+	 * @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
+	 */
+	public Identifier withPart(Identifier identifier) {
+
+		Identifier result = this;
+		for (SingleIdentifierValue part : identifier.getParts()) {
+			result = result.withPart(part.getName(), part.getValue(), part.getTargetType());
+		}
+
+		return result;
+	}
+
 	/**
 	 * Creates a new {@link Identifier} from the current instance and sets the value for {@code key}. Existing key
 	 * definitions for {@code name} are overwritten if they already exist.
@@ -188,6 +207,7 @@ public Object get(SqlIdentifier columnName) {
 		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

@@ -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,29 @@ 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 infos = 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());
+		}
+
+		Identifier identifierHolder = 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(identifier);
+		return new JdbcIdentifierBuilder(identifierHolder);
 	}
 
 	/**
@@ -62,8 +84,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;
 	}