Support AggregateReference in query derivation.

For a property of type AggregateReference one may provide an aggregate, an AggregateReference, or the id of the aggregate.

Closes #987
Original pull request: #999.

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

@@ -78,7 +78,7 @@ class AggregateReferenceConverters {
 				return null;
 			}
 
-			// if the target type is an AggregateReference we just going to assume it is of the correct type,
+			// if the target type is an AggregateReference we are going to assume it is of the correct type,
 			// because it was already converted.
 			Class<?> objectType = targetDescriptor.getObjectType();
 			if (objectType.isAssignableFrom(AggregateReference.class)) {

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

@@ -232,7 +232,8 @@ public class BasicJdbcConverter extends BasicRelationalConverter implements Jdbc
 					targetDescriptor);
 		}
 
-		if (value instanceof Array) {
+		if ( !getConversions().hasCustomReadTarget(value.getClass(), type.getType()) &&
+				value instanceof Array) {
 			try {
 				return readValue(((Array) value).getArray(), type);
 			} catch (SQLException | ConverterNotFoundException e) {

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

@@ -20,6 +20,7 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 
+import org.springframework.core.convert.ConversionService;
 import org.springframework.core.convert.converter.GenericConverter.ConvertiblePair;
 import org.springframework.core.convert.support.DefaultConversionService;
 import org.springframework.data.convert.CustomConversions;
@@ -50,6 +51,7 @@ public class JdbcCustomConversions extends CustomConversions {
 		STORE_CONVERTERS = Collections.unmodifiableCollection(converters);
 
 	}
+
 	private static final StoreConversions STORE_CONVERSIONS = StoreConversions.of(JdbcSimpleTypes.HOLDER,
 			STORE_CONVERTERS);
 

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

@@ -146,11 +146,6 @@ class JdbcQueryCreator extends RelationalQueryCreator<ParametrizedQuery> {
 			throw new IllegalArgumentException(
 					String.format("Cannot query by nested entity: %s", path.getRequiredPersistentPropertyPath().toDotPath()));
 		}
-
-		if (path.getRequiredPersistentPropertyPath().getLeafProperty().isReference()) {
-			throw new IllegalArgumentException(
-					String.format("Cannot query by reference: %s", path.getRequiredPersistentPropertyPath().toDotPath()));
-		}
 	}
 
 	/**

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

@@ -15,6 +15,7 @@
  */
 package org.springframework.data.jdbc.repository.query;
 
+import java.sql.JDBCType;
 import java.sql.Types;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -245,8 +246,8 @@ class QueryMapper {
 		return mapCondition(criteria, parameterSource, table, entity);
 	}
 
-	private Condition combine(@Nullable Condition currentCondition,
-			CriteriaDefinition.Combinator combinator, Condition nextCondition) {
+	private Condition combine(@Nullable Condition currentCondition, CriteriaDefinition.Combinator combinator,
+			Condition nextCondition) {
 
 		if (currentCondition == null) {
 			currentCondition = nextCondition;
@@ -292,6 +293,17 @@ class QueryMapper {
 
 			mappedValue = convertValue(value, propertyField.getTypeHint());
 			sqlType = propertyField.getSqlType();
+
+		} else if (propertyField instanceof MetadataBackedField //
+				&& ((MetadataBackedField) propertyField).property != null //
+				&& (criteria.getValue() == null || !criteria.getValue().getClass().isArray())) {
+
+			RelationalPersistentProperty property = ((MetadataBackedField) propertyField).property;
+			JdbcValue jdbcValue = convertSpecial(property, criteria.getValue());
+			mappedValue = jdbcValue.getValue();
+			sqlType = jdbcValue.getJdbcType() != null ? jdbcValue.getJdbcType().getVendorTypeNumber()
+					: propertyField.getSqlType();
+
 		} else {
 
 			mappedValue = convertValue(criteria.getValue(), propertyField.getTypeHint());
@@ -302,6 +314,84 @@ class QueryMapper {
 				criteria.isIgnoreCase());
 	}
 
+	/**
+	 * Converts values while taking special value types like arrays, {@link Iterable}, or {@link Pair}.
+	 * 
+	 * @param property the property to which the value relates. It determines the type to convert to. Must not be
+	 *          {@literal null}.
+	 * @param value the value to be converted.
+	 * @return a non null {@link JdbcValue} holding the converted value and the appropriate JDBC type information.
+	 */
+	private JdbcValue convertSpecial(RelationalPersistentProperty property, @Nullable Object value) {
+
+		if (value == null) {
+			return JdbcValue.of(null, JDBCType.NULL);
+		}
+
+		if (value instanceof Pair) {
+
+			final JdbcValue first = convertSimple(property, ((Pair<?, ?>) value).getFirst());
+			final JdbcValue second = convertSimple(property, ((Pair<?, ?>) value).getSecond());
+			return JdbcValue.of(Pair.of(first.getValue(), second.getValue()), first.getJdbcType());
+		}
+
+		if (value instanceof Iterable) {
+
+			List<Object> mapped = new ArrayList<>();
+			JDBCType jdbcType = null;
+
+			for (Object o : (Iterable<?>) value) {
+
+				final JdbcValue jdbcValue = convertSimple(property, o);
+				if (jdbcType == null) {
+					jdbcType = jdbcValue.getJdbcType();
+				}
+
+				mapped.add(jdbcValue.getValue());
+			}
+
+			return JdbcValue.of(mapped, jdbcType);
+		}
+
+		if (value.getClass().isArray()) {
+
+			final Object[] valueAsArray = (Object[]) value;
+			final Object[] mappedValueArray = new Object[valueAsArray.length];
+			JDBCType jdbcType = null;
+
+			for (int i = 0; i < valueAsArray.length; i++) {
+
+				final JdbcValue jdbcValue = convertSimple(property, valueAsArray[i]);
+				if (jdbcType == null) {
+					jdbcType = jdbcValue.getJdbcType();
+				}
+
+				mappedValueArray[i] = jdbcValue.getValue();
+			}
+
+			return JdbcValue.of(mappedValueArray, jdbcType);
+		}
+
+		return convertSimple(property, value);
+	}
+
+	/**
+	 * Converts values to a {@link JdbcValue}.
+	 *
+	 * @param property the property to which the value relates. It determines the type to convert to. Must not be
+	 *          {@literal null}.
+	 * @param value the value to be converted.
+	 * @return a non null {@link JdbcValue} holding the converted value and the appropriate JDBC type information.
+	 */
+	private JdbcValue convertSimple(RelationalPersistentProperty property, Object value) {
+
+		return converter.writeJdbcValue( //
+				value, //
+				converter.getColumnType(property), //
+				converter.getSqlType(property) //
+		);
+	}
+
 	private Condition mapEmbeddedObjectCondition(CriteriaDefinition criteria, MapSqlParameterSource parameterSource,
 			Table table, RelationalPersistentProperty embeddedProperty) {
 
@@ -740,11 +830,6 @@ class QueryMapper {
 				return this.property.getTypeInformation();
 			}
 
-			if (this.property.getType().isInterface()
-					|| (java.lang.reflect.Modifier.isAbstract(this.property.getType().getModifiers()))) {
-				return ClassTypeInformation.OBJECT;
-			}
-
 			return this.property.getTypeInformation();
 		}
 

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

@@ -46,7 +46,8 @@ class AggregateReferenceConvertersUnitTests {
 	@Test // GH-992
 	void convertsFromSimpleValue() {
 
-		ResolvableType aggregateReferenceWithIdTypeInteger = ResolvableType.forClassWithGenerics(AggregateReference.class, String.class, Integer.class);
+		ResolvableType aggregateReferenceWithIdTypeInteger = ResolvableType.forClassWithGenerics(AggregateReference.class,
+				String.class, Integer.class);
 		Object converted = conversionService.convert(23, TypeDescriptor.forObject(23),
 				new TypeDescriptor(aggregateReferenceWithIdTypeInteger, null, null));
 
@@ -56,7 +57,8 @@ class AggregateReferenceConvertersUnitTests {
 	@Test // GH-992
 	void convertsFromSimpleValueThatNeedsSeparateConversion() {
 
-		ResolvableType aggregateReferenceWithIdTypeInteger = ResolvableType.forClassWithGenerics(AggregateReference.class, String.class, Long.class);
+		ResolvableType aggregateReferenceWithIdTypeInteger = ResolvableType.forClassWithGenerics(AggregateReference.class,
+				String.class, Long.class);
 		Object converted = conversionService.convert(23, TypeDescriptor.forObject(23),
 				new TypeDescriptor(aggregateReferenceWithIdTypeInteger, null, null));
 

spring-data-jdbc/src/test/java/org/springframework/data/jdbc/repository/JdbcRepositoryIntegrationTests.java

@@ -50,6 +50,7 @@ import org.springframework.data.domain.Page;
 import org.springframework.data.domain.PageRequest;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Slice;
+import org.springframework.data.jdbc.core.mapping.AggregateReference;
 import org.springframework.data.jdbc.repository.query.Modifying;
 import org.springframework.data.jdbc.repository.query.Query;
 import org.springframework.data.jdbc.repository.support.JdbcRepositoryFactory;
@@ -514,6 +515,32 @@ public class JdbcRepositoryIntegrationTests {
 		assertThat(result).extracting(e -> e.idProp).containsExactly(entity.idProp);
 	}
 
+	@Test // #987
+	void queryBySimpleReference() {
+
+		final DummyEntity one = repository.save(createDummyEntity());
+		DummyEntity two = createDummyEntity();
+		two.ref = AggregateReference.to(one.idProp);
+		two = repository.save(two);
+
+		List<DummyEntity> result = repository.findByRef(one.idProp.intValue());
+
+		assertThat(result).extracting(e -> e.idProp).containsExactly(two.idProp);
+	}
+
+	@Test // #987
+	void queryByAggregateReference() {
+
+		final DummyEntity one = repository.save(createDummyEntity());
+		DummyEntity two = createDummyEntity();
+		two.ref = AggregateReference.to(one.idProp);
+		two = repository.save(two);
+
+		List<DummyEntity> result = repository.findByRef(two.ref);
+
+		assertThat(result).extracting(e -> e.idProp).containsExactly(two.idProp);
+	}
+
 	private Instant createDummyBeforeAndAfterNow() {
 
 		Instant now = Instant.now();
@@ -585,6 +612,9 @@ public class JdbcRepositoryIntegrationTests {
 		void updateWithIntervalCalculation(@Param("id") Long id, @Param("start") LocalDateTime start);
 
 		List<DummyEntity> findByFlagTrue();
+
+		List<DummyEntity> findByRef(int ref);
+		List<DummyEntity> findByRef(AggregateReference<DummyEntity, Long> ref);
 	}
 
 	@Configuration
@@ -637,6 +667,7 @@ public class JdbcRepositoryIntegrationTests {
 		OffsetDateTime offsetDateTime;
 		@Id private Long idProp;
 		boolean flag;
+		AggregateReference<DummyEntity, Long> ref;
 
 		public DummyEntity(String name) {
 			this.name = name;

spring-data-jdbc/src/test/java/org/springframework/data/jdbc/repository/query/PartTreeJdbcQueryUnitTests.java

@@ -16,6 +16,7 @@
 package org.springframework.data.jdbc.repository.query;
 
 import static org.assertj.core.api.Assertions.*;
+import static org.assertj.core.api.SoftAssertions.*;
 import static org.mockito.Mockito.*;
 
 import lombok.AllArgsConstructor;
@@ -27,11 +28,9 @@ import java.util.Date;
 import java.util.List;
 import java.util.Properties;
 
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.junit.jupiter.MockitoExtension;
-
 import org.springframework.data.annotation.Id;
 import org.springframework.data.jdbc.core.convert.BasicJdbcConverter;
 import org.springframework.data.jdbc.core.convert.JdbcConverter;
@@ -64,7 +63,7 @@ import org.springframework.jdbc.core.namedparam.NamedParameterJdbcOperations;
 public class PartTreeJdbcQueryUnitTests {
 
 	private static final String TABLE = "\"users\"";
-	private static final String ALL_FIELDS = "\"users\".\"ID\" AS \"ID\", \"users\".\"AGE\" AS \"AGE\", \"hated\".\"USER\" AS \"HATED_USER\", \"users\".\"ACTIVE\" AS \"ACTIVE\", \"users\".\"LAST_NAME\" AS \"LAST_NAME\", \"users\".\"FIRST_NAME\" AS \"FIRST_NAME\", \"users\".\"DATE_OF_BIRTH\" AS \"DATE_OF_BIRTH\", \"users\".\"HOBBY_REFERENCE\" AS \"HOBBY_REFERENCE\", \"hated\".\"NAME\" AS \"HATED_NAME\", \"users\".\"USER_CITY\" AS \"USER_CITY\", \"users\".\"USER_STREET\" AS \"USER_STREET\"";
+	private static final String ALL_FIELDS = "\"users\".\"ID\" AS \"ID\", \"users\".\"AGE\" AS \"AGE\", \"users\".\"ACTIVE\" AS \"ACTIVE\", \"users\".\"LAST_NAME\" AS \"LAST_NAME\", \"users\".\"FIRST_NAME\" AS \"FIRST_NAME\", \"users\".\"DATE_OF_BIRTH\" AS \"DATE_OF_BIRTH\", \"users\".\"HOBBY_REFERENCE\" AS \"HOBBY_REFERENCE\", \"hated\".\"NAME\" AS \"HATED_NAME\", \"users\".\"USER_CITY\" AS \"USER_CITY\", \"users\".\"USER_STREET\" AS \"USER_STREET\"";
 	private static final String JOIN_CLAUSE = "FROM \"users\" LEFT OUTER JOIN \"HOBBY\" \"hated\" ON \"hated\".\"USER\" = \"users\".\"ID\"";
 	private static final String BASE_SELECT = "SELECT " + ALL_FIELDS + " " + JOIN_CLAUSE;
 
@@ -79,11 +78,22 @@ public class PartTreeJdbcQueryUnitTests {
 		assertThatIllegalArgumentException().isThrownBy(() -> createQuery(queryMethod));
 	}
 
-	@Test // DATAJDBC-318
-	public void shouldFailForQueryByAggregateReference() throws Exception {
+	@Test // #922
+	public void createQueryByAggregateReference() throws Exception {
 
 		JdbcQueryMethod queryMethod = getQueryMethod("findAllByHobbyReference", Hobby.class);
-		assertThatIllegalArgumentException().isThrownBy(() -> createQuery(queryMethod));
+		PartTreeJdbcQuery jdbcQuery = createQuery(queryMethod);
+		final Hobby hobby = new Hobby();
+		hobby.name = "twentythree";
+		ParametrizedQuery query = jdbcQuery.createQuery(getAccessor(queryMethod, new Object[] {hobby}), returnedType);
+
+		assertSoftly(softly -> {
+
+			softly.assertThat(query.getQuery())
+					.isEqualTo(BASE_SELECT + " WHERE " + TABLE + ".\"HOBBY_REFERENCE\" = :hobby_reference");
+
+			softly.assertThat(query.getParameterSource().getValue("hobby_reference")).isEqualTo("twentythree");
+		});
 	}
 
 	@Test // DATAJDBC-318
@@ -100,11 +110,38 @@ public class PartTreeJdbcQueryUnitTests {
 		assertThatIllegalArgumentException().isThrownBy(() -> createQuery(queryMethod));
 	}
 
-	@Test // DATAJDBC-318
-	public void shouldFailForAggregateReference() throws Exception {
+	@Test // #922
+	public void createQueryForQueryByAggregateReference() throws Exception {
 
-		JdbcQueryMethod queryMethod = getQueryMethod("findByAnotherEmbeddedList", Object.class);
-		assertThatIllegalArgumentException().isThrownBy(() -> createQuery(queryMethod));
+		JdbcQueryMethod queryMethod = getQueryMethod("findViaReferenceByHobbyReference", AggregateReference.class);
+		PartTreeJdbcQuery jdbcQuery = createQuery(queryMethod);
+		final AggregateReference<Object, String> hobby = AggregateReference.to("twentythree");
+		ParametrizedQuery query = jdbcQuery.createQuery(getAccessor(queryMethod, new Object[] {hobby}), returnedType);
+
+		assertSoftly(softly -> {
+
+			softly.assertThat(query.getQuery())
+					.isEqualTo(BASE_SELECT + " WHERE " + TABLE + ".\"HOBBY_REFERENCE\" = :hobby_reference");
+
+			softly.assertThat(query.getParameterSource().getValue("hobby_reference")).isEqualTo("twentythree");
+		});
+	}
+
+	@Test // #922
+	public void createQueryForQueryByAggregateReferenceId() throws Exception {
+
+		JdbcQueryMethod queryMethod = getQueryMethod("findViaIdByHobbyReference", String.class);
+		PartTreeJdbcQuery jdbcQuery = createQuery(queryMethod);
+		final String hobby = "twentythree";
+		ParametrizedQuery query = jdbcQuery.createQuery(getAccessor(queryMethod, new Object[] {hobby}), returnedType);
+
+		assertSoftly(softly -> {
+
+			softly.assertThat(query.getQuery())
+					.isEqualTo(BASE_SELECT + " WHERE " + TABLE + ".\"HOBBY_REFERENCE\" = :hobby_reference");
+
+			softly.assertThat(query.getParameterSource().getValue("hobby_reference")).isEqualTo("twentythree");
+		});
 	}
 
 	@Test // DATAJDBC-318
@@ -176,6 +213,7 @@ public class PartTreeJdbcQueryUnitTests {
 				+ ".\"FIRST_NAME\" = :first_name)");
 	}
 
+
 	@Test // DATAJDBC-318
 	public void createsQueryToFindAllEntitiesByDateAttributeBetween() throws Exception {
 
@@ -186,11 +224,14 @@ public class PartTreeJdbcQueryUnitTests {
 		RelationalParametersParameterAccessor accessor = getAccessor(queryMethod, new Object[] { from, to });
 		ParametrizedQuery query = jdbcQuery.createQuery(accessor, returnedType);
 
-		assertThat(query.getQuery())
+		assertSoftly(softly -> {
+
+			softly.assertThat(query.getQuery())
 					.isEqualTo(BASE_SELECT + " WHERE " + TABLE + ".\"DATE_OF_BIRTH\" BETWEEN :date_of_birth AND :date_of_birth1");
 
-		assertThat(query.getParameterSource().getValue("date_of_birth")).isEqualTo(from);
-		assertThat(query.getParameterSource().getValue("date_of_birth1")).isEqualTo(to);
+			softly.assertThat(query.getParameterSource().getValue("date_of_birth")).isEqualTo(from);
+			softly.assertThat(query.getParameterSource().getValue("date_of_birth1")).isEqualTo(to);
+		});
 	}
 
 	@Test // DATAJDBC-318
@@ -250,6 +291,7 @@ public class PartTreeJdbcQueryUnitTests {
 
 	@Test // DATAJDBC-318
 	public void createsQueryToFindAllEntitiesByDateAttributeBefore() throws Exception {
+
 		JdbcQueryMethod queryMethod = getQueryMethod("findAllByDateOfBirthBefore", Date.class);
 		PartTreeJdbcQuery jdbcQuery = createQuery(queryMethod);
 		RelationalParametersParameterAccessor accessor = getAccessor(queryMethod, new Object[] { new Date() });
@@ -608,6 +650,10 @@ public class PartTreeJdbcQueryUnitTests {
 
 		List<User> findAllByHobbyReference(Hobby hobby);
 
+		List<User> findViaReferenceByHobbyReference(AggregateReference<Hobby, String> hobby);
+
+		List<User> findViaIdByHobbyReference(String hobby);
+
 		List<User> findAllByLastNameAndFirstName(String lastName, String firstName);
 
 		List<User> findAllByLastNameOrFirstName(String lastName, String firstName);
@@ -708,6 +754,7 @@ public class PartTreeJdbcQueryUnitTests {
 	}
 
 	static class Hobby {
+		@Id
 		String name;
 	}
 }

spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-db2.sql

@@ -6,5 +6,6 @@ CREATE TABLE dummy_entity
     NAME             VARCHAR(100),
     POINT_IN_TIME    TIMESTAMP,
     OFFSET_DATE_TIME TIMESTAMP, -- with time zone is only supported with z/OS
-    FLAG BOOLEAN
+    FLAG             BOOLEAN,
+    REF              BIGINT
 );

spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-h2.sql

@@ -4,5 +4,6 @@ CREATE TABLE dummy_entity
     NAME             VARCHAR(100),
     POINT_IN_TIME    TIMESTAMP,
     OFFSET_DATE_TIME TIMESTAMP WITH TIME ZONE,
-    FLAG BOOLEAN
+    FLAG             BOOLEAN,
+    REF              BIGINT
 );

spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-hsql.sql

@@ -4,5 +4,6 @@ CREATE TABLE dummy_entity
     NAME             VARCHAR(100),
     POINT_IN_TIME    TIMESTAMP,
     OFFSET_DATE_TIME TIMESTAMP WITH TIME ZONE,
-    FLAG BOOLEAN
+    FLAG             BOOLEAN,
+    REF              BIGINT
 );

spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-mariadb.sql

@@ -4,5 +4,6 @@ CREATE TABLE dummy_entity
     NAME             VARCHAR(100),
     POINT_IN_TIME    TIMESTAMP(3),
     OFFSET_DATE_TIME TIMESTAMP(3),
-    FLAG BOOLEAN
+    FLAG             BOOLEAN,
+    REF              BIGINT
 );

spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-mssql.sql

@@ -5,5 +5,6 @@ CREATE TABLE dummy_entity
     NAME             VARCHAR(100),
     POINT_IN_TIME    DATETIME,
     OFFSET_DATE_TIME DATETIMEOFFSET,
-    FLAG BIT
+    FLAG             BIT,
+    REF              BIGINT
 );

spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-mysql.sql

@@ -1,4 +1,5 @@
-SET SQL_MODE='ALLOW_INVALID_DATES';
+SET
+SQL_MODE='ALLOW_INVALID_DATES';
 
 CREATE TABLE DUMMY_ENTITY
 (
@@ -6,5 +7,6 @@ CREATE TABLE DUMMY_ENTITY
     NAME             VARCHAR(100),
     POINT_IN_TIME    TIMESTAMP(3) DEFAULT NULL,
     OFFSET_DATE_TIME TIMESTAMP(3) DEFAULT NULL,
-    FLAG BIT(1)
+    FLAG             BIT(1),
+    REF              BIGINT
 );

spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-oracle.sql

@@ -6,5 +6,6 @@ CREATE TABLE DUMMY_ENTITY
     NAME             VARCHAR2(100),
     POINT_IN_TIME    TIMESTAMP,
     OFFSET_DATE_TIME TIMESTAMP WITH TIME ZONE,
-    FLAG NUMBER(1,0)
+    FLAG             NUMBER(1,0),
+    REF              NUMBER
 );

spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-postgres.sql

@@ -5,5 +5,6 @@ CREATE TABLE dummy_entity
     NAME             VARCHAR(100),
     POINT_IN_TIME    TIMESTAMP,
     OFFSET_DATE_TIME TIMESTAMP WITH TIME ZONE,
-    FLAG BOOLEAN
+    FLAG             BOOLEAN,
+    REF              BIGINT
 );

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

@@ -157,8 +157,12 @@ public class BasicRelationalConverter implements RelationalConverter {
 			return null;
 		}
 
-		if (conversions.hasCustomReadTarget(value.getClass(), type.getType())) {
-			return conversionService.convert(value, type.getType());
+		if (getConversions().hasCustomReadTarget(value.getClass(), type.getType())) {
+
+			TypeDescriptor sourceDescriptor = TypeDescriptor.valueOf(value.getClass());
+			TypeDescriptor targetDescriptor = typeInformationToTypeDescriptor(type);
+
+			return getConversionService().convert(value, sourceDescriptor, targetDescriptor);
 		}
 
 		return getPotentiallyConvertedSimpleRead(value, type.getType());
@@ -246,6 +250,13 @@ public class BasicRelationalConverter implements RelationalConverter {
 		return conversionService.convert(value, target);
 	}
 
+	protected static TypeDescriptor typeInformationToTypeDescriptor(TypeInformation<?> type) {
+
+		Class<?>[] generics = type.getTypeArguments().stream().map(TypeInformation::getType).toArray(Class[]::new);
+
+		return new TypeDescriptor(ResolvableType.forClassWithGenerics(type.getType(), generics), null, null);
+	}
+
 	/**
 	 * Converter-aware {@link ParameterValueProvider}.
 	 *