DATAJDBC-113 - Added support for Set-valued references.

When loading the referenced entities get loaded by a separate select statement.
Changes to a collection get handled just like 1-1 relationships by deleting all and reinserting them again.

Introduced a separate method in JdbcPersistentProperty instead of just unsing the table name, although it currently still is the table name.

Now providing SqlTypes in SqlParameterSource, since MySql has a problem with null parameters when it doesn't know the type.

Fixed an error where a SQL-statement generated by the SqlGenerator was obviously wrong.

src/main/java/org/springframework/data/jdbc/core/DefaultJdbcInterpreter.java

@@ -17,7 +17,6 @@ package org.springframework.data.jdbc.core;
 
 import java.util.HashMap;
 import java.util.Map;
-import java.util.Optional;
 
 import org.springframework.data.jdbc.core.conversion.DbAction;
 import org.springframework.data.jdbc.core.conversion.DbAction.Delete;
@@ -73,8 +72,7 @@ class DefaultJdbcInterpreter implements Interpreter {
 	public <T> void interpret(Delete<T> delete) {
 
 		if (delete.getPropertyPath() == null) {
-			template.doDelete(delete.getRootId(), Optional.ofNullable(delete.getEntity()),
-					delete.getEntityType());
+			template.doDelete(delete.getRootId(), delete.getEntityType());
 		} else {
 			template.doDelete(delete.getRootId(), delete.getPropertyPath());
 		}

src/main/java/org/springframework/data/jdbc/core/EntityRowMapper.java

@@ -16,10 +16,10 @@
 package org.springframework.data.jdbc.core;
 
 import lombok.NonNull;
-import lombok.RequiredArgsConstructor;
 
 import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.util.Set;
 
 import org.springframework.core.convert.ConversionService;
 import org.springframework.data.convert.ClassGeneratingEntityInstantiator;
@@ -42,13 +42,26 @@ import org.springframework.jdbc.core.RowMapper;
  * @author Oliver Gierke
  * @since 2.0
  */
-@RequiredArgsConstructor
 class EntityRowMapper<T> implements RowMapper<T> {
 
 	private final JdbcPersistentEntity<T> entity;
 	private final EntityInstantiator instantiator = new ClassGeneratingEntityInstantiator();
 	private final ConversionService conversions;
 	private final JdbcMappingContext context;
+	private final JdbcEntityOperations template;
+	private final JdbcPersistentProperty idProperty;
+
+	@java.beans.ConstructorProperties({ "entity", "conversions", "context", "template" })
+	public EntityRowMapper(JdbcPersistentEntity<T> entity, ConversionService conversions, JdbcMappingContext context,
+			JdbcEntityOperations template) {
+
+		this.entity = entity;
+		this.conversions = conversions;
+		this.context = context;
+		this.template = template;
+
+		idProperty = entity.getRequiredIdProperty();
+	}
 
 	/*
 	 * (non-Javadoc)
@@ -58,11 +71,19 @@ class EntityRowMapper<T> implements RowMapper<T> {
 	public T mapRow(ResultSet resultSet, int rowNumber) throws SQLException {
 
 		T result = createInstance(resultSet);
-		PersistentPropertyAccessor accessor = entity.getPropertyAccessor(result);
-		ConvertingPropertyAccessor propertyAccessor = new ConvertingPropertyAccessor(accessor, conversions);
+
+		ConvertingPropertyAccessor propertyAccessor = new ConvertingPropertyAccessor(entity.getPropertyAccessor(result),
+				conversions);
+
+		Object id = readFrom(resultSet, idProperty, "");
 
 		for (JdbcPersistentProperty property : entity) {
-			propertyAccessor.setProperty(property, readFrom(resultSet, property, ""));
+
+			if (Set.class.isAssignableFrom(property.getType())) {
+				propertyAccessor.setProperty(property, template.findAllByProperty(id, property));
+			} else {
+				propertyAccessor.setProperty(property, readFrom(resultSet, property, ""));
+			}
 		}
 
 		return result;
@@ -91,7 +112,8 @@ class EntityRowMapper<T> implements RowMapper<T> {
 		String prefix = property.getName() + "_";
 
 		@SuppressWarnings("unchecked")
-		JdbcPersistentEntity<S> entity = (JdbcPersistentEntity<S>) context.getRequiredPersistentEntity(property.getType());
+		JdbcPersistentEntity<S> entity = (JdbcPersistentEntity<S>) context
+				.getRequiredPersistentEntity(property.getActualType());
 
 		if (readFrom(rs, entity.getRequiredIdProperty(), prefix) == null) {
 			return null;
@@ -109,13 +131,25 @@ class EntityRowMapper<T> implements RowMapper<T> {
 		return instance;
 	}
 
-	@RequiredArgsConstructor(staticName = "of")
 	private static class ResultSetParameterValueProvider implements ParameterValueProvider<JdbcPersistentProperty> {
 
 		@NonNull private final ResultSet resultSet;
 		@NonNull private final ConversionService conversionService;
 		@NonNull private final String prefix;
 
+		@java.beans.ConstructorProperties({ "resultSet", "conversionService", "prefix" })
+		private ResultSetParameterValueProvider(ResultSet resultSet, ConversionService conversionService, String prefix) {
+
+			this.resultSet = resultSet;
+			this.conversionService = conversionService;
+			this.prefix = prefix;
+		}
+
+		public static ResultSetParameterValueProvider of(ResultSet resultSet, ConversionService conversionService,
+				String prefix) {
+			return new ResultSetParameterValueProvider(resultSet, conversionService, prefix);
+		}
+
 		/*
 		 * (non-Javadoc)
 		 * @see org.springframework.data.mapping.model.ParameterValueProvider#getParameterValue(org.springframework.data.mapping.PreferredConstructor.Parameter)

src/main/java/org/springframework/data/jdbc/core/JdbcEntityOperations.java

@@ -17,6 +17,8 @@ package org.springframework.data.jdbc.core;
 
 import java.util.Map;
 
+import org.springframework.data.jdbc.mapping.model.JdbcPersistentProperty;
+
 /**
  * Specifies a operations one can perform on a database, based on an <em>Domain Type</em>.
  *
@@ -44,5 +46,8 @@ public interface JdbcEntityOperations {
 
 	<T> Iterable<T> findAll(Class<T> domainType);
 
+	<T> Iterable<T> findAllByProperty(Object id, JdbcPersistentProperty property);
+
 	<T> boolean existsById(Object id, Class<T> domainType);
+
 }

src/main/java/org/springframework/data/jdbc/core/JdbcEntityTemplate.java

@@ -15,8 +15,6 @@
  */
 package org.springframework.data.jdbc.core;
 
-import java.math.BigInteger;
-import java.sql.Types;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
@@ -46,10 +44,10 @@ import org.springframework.data.jdbc.mapping.model.JdbcMappingContext;
 import org.springframework.data.jdbc.mapping.model.JdbcPersistentEntity;
 import org.springframework.data.jdbc.mapping.model.JdbcPersistentEntityInformation;
 import org.springframework.data.jdbc.mapping.model.JdbcPersistentProperty;
+import org.springframework.data.jdbc.support.JdbcUtil;
 import org.springframework.data.mapping.PropertyHandler;
 import org.springframework.data.mapping.PropertyPath;
 import org.springframework.data.repository.core.EntityInformation;
-import org.springframework.jdbc.core.SqlParameterValue;
 import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
 import org.springframework.jdbc.core.namedparam.NamedParameterJdbcOperations;
 import org.springframework.jdbc.support.GeneratedKeyHolder;
@@ -101,7 +99,8 @@ public class JdbcEntityTemplate implements JdbcEntityOperations {
 	@Override
 	public <T> void save(T instance, Class<T> domainType) {
 
-		JdbcPersistentEntityInformation<T, ?> entityInformation = context.getRequiredPersistentEntityInformation(domainType);
+		JdbcPersistentEntityInformation<T, ?> entityInformation = context
+				.getRequiredPersistentEntityInformation(domainType);
 
 		AggregateChange change = createChange(instance);
 
@@ -128,16 +127,17 @@ public class JdbcEntityTemplate implements JdbcEntityOperations {
 		JdbcPersistentEntityInformation<T, ?> entityInformation = context
 				.getRequiredPersistentEntityInformation(domainType);
 
-		Map<String, Object> propertyMap = getPropertyMap(instance, persistentEntity);
+		MapSqlParameterSource parameterSource = getPropertyMap(instance, persistentEntity);
 
 		Object idValue = getIdValueOrNull(instance, persistentEntity);
 		JdbcPersistentProperty idProperty = persistentEntity.getRequiredIdProperty();
-		propertyMap.put(idProperty.getColumnName(), convert(idValue, idProperty.getColumnType()));
+		parameterSource.addValue(idProperty.getColumnName(), convert(idValue, idProperty.getColumnType()),
+				JdbcUtil.sqlTypeFor(idProperty.getColumnType()));
 
-		propertyMap.putAll(additionalParameters);
+		additionalParameters.forEach(parameterSource::addValue);
 
-		operations.update(sql(domainType).getInsert(idValue == null, additionalParameters.keySet()),
-				new MapSqlParameterSource(propertyMap), holder);
+		operations.update(sql(domainType).getInsert(idValue == null, additionalParameters.keySet()), parameterSource,
+				holder);
 
 		setIdFromJdbc(instance, holder, persistentEntity);
 
@@ -198,6 +198,17 @@ public class JdbcEntityTemplate implements JdbcEntityOperations {
 		return operations.query(findAllInListSql, parameter, getEntityRowMapper(domainType));
 	}
 
+	@Override
+	public <T> Iterable<T> findAllByProperty(Object id, JdbcPersistentProperty property) {
+
+		Class<?> actualType = property.getActualType();
+		String findAllByProperty = sql(actualType).getFindAllByProperty(property.getReverseColumnName());
+
+		MapSqlParameterSource parameter = new MapSqlParameterSource(property.getReverseColumnName(), id);
+
+		return (Iterable<T>) operations.query(findAllByProperty, parameter, getEntityRowMapper(actualType));
+	}
+
 	@Override
 	public <S> void delete(S entity, Class<S> domainType) {
 
@@ -247,7 +258,7 @@ public class JdbcEntityTemplate implements JdbcEntityOperations {
 
 	}
 
-	void doDelete(Object id, Optional<Object> optionalEntity, Class<?> domainType) {
+	void doDelete(Object id, Class<?> domainType) {
 
 		String deleteByIdSql = sql(domainType).getDeleteById();
 		MapSqlParameterSource parameter = createIdParameterSource(id, domainType);
@@ -281,20 +292,16 @@ public class JdbcEntityTemplate implements JdbcEntityOperations {
 				convert(id, getRequiredPersistentEntity(domainType).getRequiredIdProperty().getColumnType()));
 	}
 
-	private <S> Map<String, Object> getPropertyMap(final S instance, JdbcPersistentEntity<S> persistentEntity) {
+	private <S> MapSqlParameterSource getPropertyMap(final S instance, JdbcPersistentEntity<S> persistentEntity) {
 
-		Map<String, Object> parameters = new HashMap<>();
+		MapSqlParameterSource parameters = new MapSqlParameterSource();
 
 		persistentEntity.doWithProperties((PropertyHandler<JdbcPersistentProperty>) property -> {
 			if (!property.isEntity()) {
 				Object value = persistentEntity.getPropertyAccessor(instance).getProperty(property);
 
 				Object convertedValue = convert(value, property.getColumnType());
-				if (convertedValue instanceof BigInteger) {
-					parameters.put(property.getColumnName(), new SqlParameterValue(Types.BIGINT, convertedValue));
-				} else {
-					parameters.put(property.getColumnName(), convertedValue);
-				}
+				parameters.addValue(property.getColumnName(), convertedValue, JdbcUtil.sqlTypeFor(property.getColumnType()));
 			}
 		});
 
@@ -372,7 +379,7 @@ public class JdbcEntityTemplate implements JdbcEntityOperations {
 	}
 
 	private <T> EntityRowMapper<T> getEntityRowMapper(Class<T> domainType) {
-		return new EntityRowMapper<>(getRequiredPersistentEntity(domainType), conversions, context);
+		return new EntityRowMapper<>(getRequiredPersistentEntity(domainType), conversions, context, this);
 	}
 
 	<T> void doDeleteAll(Class<T> domainType, PropertyPath propertyPath) {
@@ -381,4 +388,8 @@ public class JdbcEntityTemplate implements JdbcEntityOperations {
 				.update(sql(propertyPath == null ? domainType : propertyPath.getOwningType().getType())
 						.createDeleteAllSql(propertyPath));
 	}
+
+	public NamedParameterJdbcOperations getOperations() {
+		return operations;
+	}
 }

src/main/java/org/springframework/data/jdbc/core/SqlGenerator.java

@@ -16,6 +16,7 @@
 package org.springframework.data.jdbc.core;
 
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.List;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -58,8 +59,7 @@ class SqlGenerator {
 	private final Lazy<String> deleteByListSql = Lazy.of(this::createDeleteByListSql);
 	private final SqlGeneratorSource sqlGeneratorSource;
 
-	SqlGenerator(JdbcMappingContext context, JdbcPersistentEntity<?> entity,
-				 SqlGeneratorSource sqlGeneratorSource) {
+	SqlGenerator(JdbcMappingContext context, JdbcPersistentEntity<?> entity, SqlGeneratorSource sqlGeneratorSource) {
 
 		this.context = context;
 		this.entity = entity;
@@ -88,6 +88,10 @@ class SqlGenerator {
 		return findAllSql.get();
 	}
 
+	String getFindAllByProperty(String columnName) {
+		return String.format("%s WHERE %s = :%s", findAllSql.get(), columnName, columnName);
+	}
+
 	String getExists() {
 		return existsSql.get();
 	}
@@ -96,7 +100,7 @@ class SqlGenerator {
 		return findOneSql.get();
 	}
 
-	String getInsert(boolean excludeId, Set additionalColumns) {
+	String getInsert(boolean excludeId, Set<String> additionalColumns) {
 		return createInsertSql(excludeId, additionalColumns);
 	}
 
@@ -124,39 +128,51 @@ class SqlGenerator {
 	}
 
 	private SelectBuilder createSelectBuilder() {
+
 		SelectBuilder builder = new SelectBuilder(entity.getTableName());
-
-		for (JdbcPersistentProperty property : entity) {
-			if (!property.isEntity()) {
-
-				builder.column(cb -> cb //
-						.tableAlias(entity.getTableName()) //
-						.column(property.getColumnName()) //
-						.as(property.getColumnName()));
-			}
-		}
-
-		for (JdbcPersistentProperty property : entity) {
-			if (property.isEntity()) {
-
-				JdbcPersistentEntity<?> refEntity = context.getRequiredPersistentEntity(property.getType());
-				String joinAlias = property.getName();
-				builder.join(jb -> jb.leftOuter().table(refEntity.getTableName()).as(joinAlias) //
-						.where(entity.getTableName()).eq().column(entity.getTableName(), entity.getIdColumn()));
-
-				for (JdbcPersistentProperty refProperty : refEntity) {
-					builder.column( //
-							cb -> cb.tableAlias(joinAlias) //
-									.column(refProperty.getColumnName()) //
-									.as(joinAlias + "_" + refProperty.getColumnName()) //
-					);
-				}
-			}
-		}
+		addColumnsForSimpleProperties(builder);
+		addColumnsAndJoinsForOneToOneReferences(builder);
 
 		return builder;
 	}
 
+	private void addColumnsAndJoinsForOneToOneReferences(SelectBuilder builder) {
+
+		for (JdbcPersistentProperty property : entity) {
+			if (!property.isEntity() || Collection.class.isAssignableFrom(property.getType())) {
+				continue;
+			}
+
+			JdbcPersistentEntity<?> refEntity = context.getRequiredPersistentEntity(property.getActualType());
+			String joinAlias = property.getName();
+			builder.join(jb -> jb.leftOuter().table(refEntity.getTableName()).as(joinAlias) //
+					.where(property.getReverseColumnName()).eq().column(entity.getTableName(), entity.getIdColumn()));
+
+			for (JdbcPersistentProperty refProperty : refEntity) {
+				builder.column( //
+						cb -> cb.tableAlias(joinAlias) //
+								.column(refProperty.getColumnName()) //
+								.as(joinAlias + "_" + refProperty.getColumnName()) //
+				);
+			}
+		}
+	}
+
+	private void addColumnsForSimpleProperties(SelectBuilder builder) {
+
+		for (JdbcPersistentProperty property : entity) {
+
+			if (property.isEntity()) {
+				continue;
+			}
+
+			builder.column(cb -> cb //
+					.tableAlias(entity.getTableName()) //
+					.column(property.getColumnName()) //
+					.as(property.getColumnName()));
+		}
+	}
+
 	private Stream<String> getColumnNameStream(String prefix) {
 
 		return StreamUtils.createStreamFromIterator(entity.iterator()) //
@@ -191,11 +207,13 @@ class SqlGenerator {
 		return String.format("select count(*) from %s", entity.getTableName());
 	}
 
-	private String createInsertSql(boolean excludeId, Set additionalColumns) {
+	private String createInsertSql(boolean excludeId, Set<String> additionalColumns) {
 
 		String insertTemplate = "insert into %s (%s) values (%s)";
+
 		List<String> propertyNamesForInsert = new ArrayList<>(excludeId ? nonIdPropertyNames : propertyNames);
 		propertyNamesForInsert.addAll(additionalColumns);
+
 		String tableColumns = String.join(", ", propertyNamesForInsert);
 		String parameterNames = propertyNamesForInsert.stream().collect(Collectors.joining(", :", ":", ""));
 
@@ -225,29 +243,31 @@ class SqlGenerator {
 
 		JdbcPersistentEntity<?> entityToDelete = context.getRequiredPersistentEntity(PropertyPaths.getLeafType(path));
 
-		String innerMostCondition = String.format("%s IS NOT NULL", entity.getTableName(), entity.getTableName(),
-				entity.getIdColumn());
+		JdbcPersistentEntity<?> owningEntity = context.getRequiredPersistentEntity(path.getOwningType());
+		JdbcPersistentProperty property = owningEntity.getRequiredPersistentProperty(path.getSegment());
+
+		String innerMostCondition = String.format("%s IS NOT NULL", property.getReverseColumnName());
 
 		String condition = cascadeConditions(innerMostCondition, path.next());
 
-		return String.format("DELETE FROM %s WHERE %s", entityToDelete.getTableName(), condition, condition);
-
+		return String.format("DELETE FROM %s WHERE %s", entityToDelete.getTableName(), condition);
 	}
 
 	private String createDeleteByListSql() {
-		return String.format("doDelete from %s where %s in (:ids)", entity.getTableName(), entity.getIdColumn());
+		return String.format("DELETE FROM %s WHERE %s IN (:ids)", entity.getTableName(), entity.getIdColumn());
 	}
 
 	String createDeleteByPath(PropertyPath path) {
 
 		JdbcPersistentEntity<?> entityToDelete = context.getRequiredPersistentEntity(PropertyPaths.getLeafType(path));
+		JdbcPersistentEntity<?> owningEntity = context.getRequiredPersistentEntity(path.getOwningType());
+		JdbcPersistentProperty property = owningEntity.getRequiredPersistentProperty(path.getSegment());
 
-		String innerMostCondition = String.format("%s = :rootId", entity.getTableName());
+		String innerMostCondition = String.format("%s = :rootId", property.getReverseColumnName());
 
 		String condition = cascadeConditions(innerMostCondition, path.next());
 
 		return String.format("DELETE FROM %s WHERE %s", entityToDelete.getTableName(), condition);
-
 	}
 
 	private String cascadeConditions(String innerCondition, PropertyPath path) {
@@ -261,10 +281,10 @@ class SqlGenerator {
 
 		Assert.notNull(property, "could not find property for path " + path.getSegment() + " in " + entity);
 
-		String tableName = entity.getTableName();
-		String idColumn = entity.getIdColumn();
-
-		return String.format("%s IN (SELECT %s FROM %s WHERE %s)", tableName, idColumn, tableName, innerCondition);
-
+		return String.format("%s IN (SELECT %s FROM %s WHERE %s)", //
+				property.getReverseColumnName(), //
+				entity.getIdColumn(), //
+				entity.getTableName(), innerCondition //
+		);
 	}
 }

src/main/java/org/springframework/data/jdbc/core/conversion/JdbcEntityWriter.java

@@ -15,6 +15,7 @@
  */
 package org.springframework.data.jdbc.core.conversion;
 
+import java.util.Collection;
 import java.util.stream.Stream;
 
 import org.springframework.data.jdbc.core.conversion.DbAction.Insert;
@@ -26,6 +27,7 @@ import org.springframework.data.jdbc.mapping.model.JdbcPersistentProperty;
 import org.springframework.data.mapping.PersistentProperty;
 import org.springframework.data.mapping.PersistentPropertyAccessor;
 import org.springframework.data.util.StreamUtils;
+import org.springframework.util.ClassUtils;
 
 /**
  * Converts an entity that is about to be saved into {@link DbAction}s inside a {@link AggregateChange} that need to be
@@ -46,8 +48,6 @@ public class JdbcEntityWriter extends JdbcEntityWriterSupport {
 
 	private void write(Object o, AggregateChange aggregateChange, DbAction dependingOn) {
 
-		JdbcPersistentEntity<?> persistentEntity = context.getRequiredPersistentEntity(aggregateChange.getEntityType());
-
 		JdbcPersistentEntityInformation<Object, ?> entityInformation = context
 				.getRequiredPersistentEntityInformation((Class<Object>) o.getClass());
 
@@ -70,22 +70,34 @@ public class JdbcEntityWriter extends JdbcEntityWriterSupport {
 
 	private void saveReferencedEntities(Object o, AggregateChange aggregateChange, DbAction dependingOn) {
 
-		DbAction action = saveAction(o, dependingOn);
-		aggregateChange.addAction(action);
+		saveActions(o, dependingOn).forEach(a -> {
+
+			aggregateChange.addAction(a);
+			referencedEntities(o).forEach(e -> saveReferencedEntities(e, aggregateChange, a));
+		});
 
-		referencedEntities(o).forEach(e -> saveReferencedEntities(e, aggregateChange, action));
 	}
 
-	private <T> DbAction saveAction(T t, DbAction dependingOn) {
+	private <T> Stream<DbAction> saveActions(T t, DbAction dependingOn) {
+
+		if (Collection.class.isAssignableFrom(ClassUtils.getUserClass(t))) {
+			return collectionSaveAction((Collection) t, dependingOn);
+		}
+
+		return Stream.of(singleSaveAction(t, dependingOn));
+	}
+
+	private Stream<DbAction> collectionSaveAction(Collection collection, DbAction dependingOn) {
+
+		return collection.stream().map(e -> singleSaveAction(e, dependingOn));
+	}
+
+	private <T> DbAction singleSaveAction(T t, DbAction dependingOn) {
 
 		JdbcPersistentEntityInformation<T, ?> entityInformation = context
-				.getRequiredPersistentEntityInformation((Class<T>) t.getClass());
+				.getRequiredPersistentEntityInformation((Class<T>) ClassUtils.getUserClass(t));
 
-		if (entityInformation.isNew(t)) {
-			return DbAction.insert(t, dependingOn);
-		} else {
-			return DbAction.update(t, dependingOn);
-		}
+		return entityInformation.isNew(t) ? DbAction.insert(t, dependingOn) : DbAction.update(t, dependingOn);
 	}
 
 	private void insertReferencedEntities(Object o, AggregateChange aggregateChange, DbAction dependingOn) {
@@ -105,14 +117,34 @@ public class JdbcEntityWriter extends JdbcEntityWriterSupport {
 
 	private Stream<?> referencedEntity(JdbcPersistentProperty p, PersistentPropertyAccessor propertyAccessor) {
 
-		Class<?> type = p.getActualType();
+		Class<?> actualType = p.getActualType();
 		JdbcPersistentEntity<?> persistentEntity = context //
-				.getPersistentEntity(type);
+				.getPersistentEntity(actualType);
 
 		if (persistentEntity == null) {
 			return Stream.empty();
 		}
 
+		Class<?> type = p.getType();
+		if (Collection.class.isAssignableFrom(type))
+			return collectionPropertyAsStream(p, propertyAccessor);
+
+		return singlePropertyAsStream(p, propertyAccessor);
+	}
+
+	private Stream<Object> collectionPropertyAsStream(JdbcPersistentProperty p,
+			PersistentPropertyAccessor propertyAccessor) {
+
+		Object property = propertyAccessor.getProperty(p);
+		if (property == null) {
+			return Stream.empty();
+		}
+
+		return ((Collection<Object>) property).stream();
+	}
+
+	private Stream<Object> singlePropertyAsStream(JdbcPersistentProperty p, PersistentPropertyAccessor propertyAccessor) {
+
 		Object property = propertyAccessor.getProperty(p);
 		if (property == null) {
 			return Stream.empty();

src/main/java/org/springframework/data/jdbc/mapping/model/BasicJdbcPersistentProperty.java

@@ -92,9 +92,19 @@ public class BasicJdbcPersistentProperty extends AnnotationBasedPersistentProper
 	@Override
 	public Class getColumnType() {
 
-		Class columnType = columnTypeIfEntity(getType());
+		Class columnType = columnTypeIfEntity(getActualType());
 
-		return columnType == null ? columnTypeForNonEntity(getType()) : columnType;
+		return columnType == null ? columnTypeForNonEntity(getActualType()) : columnType;
+	}
+
+	@Override
+	public JdbcPersistentEntity<?> getOwner() {
+		return (JdbcPersistentEntity<?>) super.getOwner();
+	}
+
+	@Override
+	public String getReverseColumnName() {
+		return getOwner().getTableName();
 	}
 
 	private Class columnTypeIfEntity(Class type) {

src/main/java/org/springframework/data/jdbc/mapping/model/JdbcPersistentProperty.java

@@ -39,4 +39,9 @@ public interface JdbcPersistentProperty extends PersistentProperty<JdbcPersisten
 	 * @return a {@link Class} that is suitable for usage with JDBC drivers
 	 */
 	Class<?> getColumnType();
+
+	@Override
+	JdbcPersistentEntity<?> getOwner();
+
+	String getReverseColumnName();
 }

src/main/java/org/springframework/data/jdbc/support/JdbcUtil.java

@@ -0,0 +1,73 @@
+/*
+ * Copyright 2017 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
+ *
+ *      http://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.jdbc.support;
+
+import lombok.experimental.UtilityClass;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.sql.Date;
+import java.sql.Time;
+import java.sql.Timestamp;
+import java.sql.Types;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.springframework.jdbc.support.JdbcUtils;
+
+/**
+ * Contains methods dealing with the quirks of JDBC, independent of any Entity, Aggregate or Repository abstraction.
+ *
+ * @author Jens Schauder
+ */
+@UtilityClass
+public class JdbcUtil {
+
+	private static final Map<Class, Integer> sqlTypeMappings = new HashMap<>();
+
+	static {
+
+		sqlTypeMappings.put(String.class, Types.VARCHAR);
+		sqlTypeMappings.put(BigInteger.class, Types.BIGINT);
+		sqlTypeMappings.put(BigDecimal.class, Types.NUMERIC);
+		sqlTypeMappings.put(Byte.class, Types.TINYINT);
+		sqlTypeMappings.put(byte.class, Types.TINYINT);
+		sqlTypeMappings.put(Short.class, Types.SMALLINT);
+		sqlTypeMappings.put(short.class, Types.SMALLINT);
+		sqlTypeMappings.put(Integer.class, Types.INTEGER);
+		sqlTypeMappings.put(int.class, Types.INTEGER);
+		sqlTypeMappings.put(Long.class, Types.BIGINT);
+		sqlTypeMappings.put(long.class, Types.BIGINT);
+		sqlTypeMappings.put(Double.class, Types.DOUBLE);
+		sqlTypeMappings.put(double.class, Types.DOUBLE);
+		sqlTypeMappings.put(Float.class, Types.REAL);
+		sqlTypeMappings.put(float.class, Types.REAL);
+		sqlTypeMappings.put(Boolean.class, Types.BIT);
+		sqlTypeMappings.put(boolean.class, Types.BIT);
+		sqlTypeMappings.put(byte[].class, Types.VARBINARY);
+		sqlTypeMappings.put(Date.class, Types.DATE);
+		sqlTypeMappings.put(Time.class, Types.TIME);
+		sqlTypeMappings.put(Timestamp.class, Types.TIMESTAMP);
+	}
+
+	public static int sqlTypeFor(Class type) {
+		return sqlTypeMappings.keySet().stream() //
+				.filter(k -> k.isAssignableFrom(type)) //
+				.findFirst() //
+				.map(sqlTypeMappings::get) //
+				.orElse(JdbcUtils.TYPE_UNKNOWN);
+	}
+}