Support AggregateReferences with custom id type.

Restructured reading conversion process into:
- converting technology base types (JDBC Arrays).
- standard and custom conversions.
- module specific conversions (AggregateReference).

Closes #1828
Original pull request #2062

Author: schauder

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

@@ -27,9 +27,8 @@
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.springframework.context.ApplicationContextAware;
-import org.springframework.core.convert.ConverterNotFoundException;
 import org.springframework.core.convert.converter.Converter;
-import org.springframework.core.convert.converter.ConverterRegistry;
+import org.springframework.dao.NonTransientDataAccessException;
 import org.springframework.data.convert.CustomConversions;
 import org.springframework.data.jdbc.core.mapping.AggregateReference;
 import org.springframework.data.jdbc.core.mapping.JdbcValue;
@@ -91,8 +90,6 @@ public MappingJdbcConverter(RelationalMappingContext context, RelationResolver r
 
 		this.typeFactory = JdbcTypeFactory.unsupported();
 		this.relationResolver = relationResolver;
-
-		registerAggregateReferenceConverters();
 	}
 
 	/**
@@ -112,14 +109,6 @@ public MappingJdbcConverter(RelationalMappingContext context, RelationResolver r
 
 		this.typeFactory = typeFactory;
 		this.relationResolver = relationResolver;
-
-		registerAggregateReferenceConverters();
-	}
-
-	private void registerAggregateReferenceConverters() {
-
-		ConverterRegistry registry = (ConverterRegistry) getConversionService();
-		AggregateReferenceConverters.getConvertersToRegister(getConversionService()).forEach(registry::addConverter);
 	}
 
 	@Nullable
@@ -184,34 +173,78 @@ private Class<?> doGetColumnType(RelationalPersistentProperty property) {
 		return componentColumnType;
 	}
 
+	/**
+	 * Read and convert a single value that is coming from a database to the {@literal targetType} expected by the domain
+	 * model.
+	 *
+	 * @param value a value as it is returned by the driver accessing the persistence store. May be {@code null}.
+	 * @param targetType {@link TypeInformation} into which the value is to be converted. Must not be {@code null}.
+	 * @return
+	 */
 	@Override
 	@Nullable
-	public Object readValue(@Nullable Object value, TypeInformation<?> type) {
+	public Object readValue(@Nullable Object value, TypeInformation<?> targetType) {
 
-		if (value == null) {
-			return value;
+		if (null == value) {
+			return null;
 		}
 
+		TypeInformation<?> originalTargetType = targetType;
+		value = readJdbcArray(value);
+		targetType = determineNestedTargetType(targetType);
+
+		return possiblyReadToAggregateReference(getPotentiallyConvertedSimpleRead(value, targetType), originalTargetType);
+	}
+
+	/**
+	 * Unwrap a Jdbc array, if such a value is provided
+	 */
+	private Object readJdbcArray(Object value) {
+
 		if (value instanceof Array array) {
 			try {
-				return super.readValue(array.getArray(), type);
-			} catch (SQLException | ConverterNotFoundException e) {
-				LOG.info("Failed to extract a value of type %s from an Array; Attempting to use standard conversions", e);
+				return array.getArray();
+			} catch (SQLException e) {
+				throw new FailedToAccessJdbcArrayException(e);
 			}
 		}
 
-		return super.readValue(value, type);
+		return value;
+	}
+
+	/**
+	 * Determine the id type of an {@link AggregateReference} that the rest of the conversion infrastructure needs to use
+	 * as a conversion target.
+	 */
+	private TypeInformation<?> determineNestedTargetType(TypeInformation<?> ultimateTargetType) {
+
+		if (AggregateReference.class.isAssignableFrom(ultimateTargetType.getType())) {
+			// the id type of a AggregateReference
+			return ultimateTargetType.getTypeArguments().get(1);
+		}
+		return ultimateTargetType;
+	}
+
+	/**
+	 * Convert value to an {@link AggregateReference} if that is specified by the parameter targetType.
+	 */
+	private Object possiblyReadToAggregateReference(Object value, TypeInformation<?> targetType) {
+
+		if (AggregateReference.class.isAssignableFrom(targetType.getType())) {
+			return AggregateReference.to(value);
+		}
+		return value;
 	}
 
-	@Override
 	@Nullable
-	public Object writeValue(@Nullable Object value, TypeInformation<?> type) {
+	@Override
+	protected Object getPotentiallyConvertedSimpleWrite(Object value, TypeInformation<?> type) {
 
-		if (value == null) {
-			return null;
+		if (value instanceof AggregateReference<?, ?> aggregateReference) {
+			return writeValue(aggregateReference.getId(), type);
 		}
 
-		return super.writeValue(value, type);
+		return super.getPotentiallyConvertedSimpleWrite(value, type);
 	}
 
 	private boolean canWriteAsJdbcValue(@Nullable Object value) {
@@ -244,28 +277,37 @@ private boolean canWriteAsJdbcValue(@Nullable Object value) {
 	public JdbcValue writeJdbcValue(@Nullable Object value, TypeInformation<?> columnType, SQLType sqlType) {
 
 		TypeInformation<?> targetType = canWriteAsJdbcValue(value) ? TypeInformation.of(JdbcValue.class) : columnType;
+		
+		if (value instanceof AggregateReference<?, ?> aggregateReference) {
+			return writeJdbcValue(aggregateReference.getId(), columnType, sqlType);
+		}
+		
 		Object convertedValue = writeValue(value, targetType);
 
 		if (convertedValue instanceof JdbcValue result) {
 			return result;
 		}
 
-		if (convertedValue == null || !convertedValue.getClass().isArray()) {
-			return JdbcValue.of(convertedValue, sqlType);
+		if (convertedValue == null) {
+			return JdbcValue.of(null, sqlType);
 		}
 
-		Class<?> componentType = convertedValue.getClass().getComponentType();
-		if (componentType != byte.class && componentType != Byte.class) {
+		if (convertedValue.getClass().isArray()) {// array conversion
+			Class<?> componentType = convertedValue.getClass().getComponentType();
+			if (componentType != byte.class && componentType != Byte.class) {
 
-			Object[] objectArray = requireObjectArray(convertedValue);
-			return JdbcValue.of(typeFactory.createArray(objectArray), JDBCType.ARRAY);
-		}
+				Object[] objectArray = requireObjectArray(convertedValue);
+				return JdbcValue.of(typeFactory.createArray(objectArray), JDBCType.ARRAY);
+			}
 
-		if (componentType == Byte.class) {
-			convertedValue = ArrayUtils.toPrimitive((Byte[]) convertedValue);
-		}
+			if (componentType == Byte.class) {
+				convertedValue = ArrayUtils.toPrimitive((Byte[]) convertedValue);
+			}
 
-		return JdbcValue.of(convertedValue, JDBCType.BINARY);
+			return JdbcValue.of(convertedValue, JDBCType.BINARY);
+		}
+		
+		return JdbcValue.of(convertedValue, sqlType);
 	}
 
 	@SuppressWarnings("unchecked")
@@ -298,6 +340,12 @@ protected RelationalPropertyValueProvider newValueProvider(RowDocumentAccessor d
 		return super.newValueProvider(documentAccessor, evaluator, context);
 	}
 
+	private static class FailedToAccessJdbcArrayException extends NonTransientDataAccessException {
+		public FailedToAccessJdbcArrayException(SQLException e) {
+			super("Failed to read array", e);
+		}
+	}
+
 	/**
 	 * {@link RelationalPropertyValueProvider} using a resolving context to lookup relations. This is highly
 	 * context-sensitive. Note that the identifier is held here because of a chicken and egg problem, while