#30 - Add custom conversion support.

We now support custom conversions via R2dbcCustomConversions.
Custom conversions introduces simple types that depend on the used dialect.
Custom conversions and simple types are held in RelationalConverter and MappingContext.

Simple types and conversions are used by DatabaseClient and repository support to properly apply registered converters and support native types such as array-columns.

Related tickets: #22, #26.
Original pull request: #31.

src/main/java/org/springframework/data/r2dbc/config/AbstractR2dbcConfiguration.java

@@ -17,15 +17,20 @@ package org.springframework.data.r2dbc.config;
 
 import io.r2dbc.spi.ConnectionFactory;
 
+import java.util.Collections;
 import java.util.Optional;
 
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.core.convert.converter.Converter;
+import org.springframework.data.convert.CustomConversions;
+import org.springframework.data.convert.CustomConversions.StoreConversions;
 import org.springframework.data.r2dbc.dialect.Database;
 import org.springframework.data.r2dbc.dialect.Dialect;
 import org.springframework.data.r2dbc.function.DatabaseClient;
 import org.springframework.data.r2dbc.function.DefaultReactiveDataAccessStrategy;
 import org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy;
+import org.springframework.data.r2dbc.function.convert.R2dbcCustomConversions;
 import org.springframework.data.r2dbc.support.R2dbcExceptionTranslator;
 import org.springframework.data.r2dbc.support.SqlErrorCodeR2dbcExceptionTranslator;
 import org.springframework.data.relational.core.conversion.BasicRelationalConverter;
@@ -95,15 +100,21 @@ public abstract class AbstractR2dbcConfiguration {
 	 * Register a {@link RelationalMappingContext} and apply an optional {@link NamingStrategy}.
 	 *
 	 * @param namingStrategy optional {@link NamingStrategy}. Use {@link NamingStrategy#INSTANCE} as fallback.
+	 * @param r2dbcCustomConversions customized R2DBC conversions.
 	 * @return must not be {@literal null}.
 	 * @throws IllegalArgumentException if any of the required args is {@literal null}.
 	 */
 	@Bean
-	public RelationalMappingContext r2dbcMappingContext(Optional<NamingStrategy> namingStrategy) {
+	public RelationalMappingContext r2dbcMappingContext(Optional<NamingStrategy> namingStrategy,
+			R2dbcCustomConversions r2dbcCustomConversions) {
 
 		Assert.notNull(namingStrategy, "NamingStrategy must not be null!");
 
-		return new RelationalMappingContext(namingStrategy.orElse(NamingStrategy.INSTANCE));
+		RelationalMappingContext relationalMappingContext = new RelationalMappingContext(
+				namingStrategy.orElse(NamingStrategy.INSTANCE));
+		relationalMappingContext.setSimpleTypeHolder(r2dbcCustomConversions.getSimpleTypeHolder());
+
+		return relationalMappingContext;
 	}
 
 	/**
@@ -111,17 +122,37 @@ public abstract class AbstractR2dbcConfiguration {
 	 * RelationalMappingContext}.
 	 *
 	 * @param mappingContext the configured {@link RelationalMappingContext}.
+	 * @param r2dbcCustomConversions customized R2DBC conversions.
 	 * @return must not be {@literal null}.
-	 * @see #r2dbcMappingContext(Optional)
+	 * @see #r2dbcMappingContext(Optional, R2dbcCustomConversions)
 	 * @see #getDialect(ConnectionFactory)
 	 * @throws IllegalArgumentException if any of the {@literal mappingContext} is {@literal null}.
 	 */
 	@Bean
-	public ReactiveDataAccessStrategy reactiveDataAccessStrategy(RelationalMappingContext mappingContext) {
+	public ReactiveDataAccessStrategy reactiveDataAccessStrategy(RelationalMappingContext mappingContext,
+			R2dbcCustomConversions r2dbcCustomConversions) {
 
 		Assert.notNull(mappingContext, "MappingContext must not be null!");
-		return new DefaultReactiveDataAccessStrategy(getDialect(connectionFactory()),
-				new BasicRelationalConverter(mappingContext));
+
+		BasicRelationalConverter converter = new BasicRelationalConverter(mappingContext, r2dbcCustomConversions);
+
+		return new DefaultReactiveDataAccessStrategy(getDialect(connectionFactory()), converter);
+	}
+
+	/**
+	 * Register custom {@link Converter}s in a {@link CustomConversions} object if required. These
+	 * {@link CustomConversions} will be registered with the {@link BasicRelationalConverter} and
+	 * {@link #r2dbcMappingContext(Optional, R2dbcCustomConversions)}. Returns an empty {@link R2dbcCustomConversions}
+	 * instance by default.
+	 *
+	 * @return must not be {@literal null}.
+	 */
+	@Bean
+	public R2dbcCustomConversions r2dbcCustomConversions() {
+
+		Dialect dialect = getDialect(connectionFactory());
+		StoreConversions storeConversions = StoreConversions.of(dialect.getSimpleTypeHolder());
+		return new R2dbcCustomConversions(storeConversions, Collections.emptyList());
 	}
 
 	/**

src/main/java/org/springframework/data/r2dbc/dialect/Dialect.java

@@ -1,5 +1,11 @@
 package org.springframework.data.r2dbc.dialect;
 
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+
+import org.springframework.data.mapping.model.SimpleTypeHolder;
+
 /**
  * Represents a dialect that is implemented by a particular database.
  *
@@ -26,10 +32,41 @@ public interface Dialect {
 	@Deprecated
 	String generatedKeysClause();
 
+	/**
+	 * Return a collection of types that are natively supported by this database/driver. Defaults to
+	 * {@link Collections#emptySet()}.
+	 *
+	 * @return a collection of types that are natively supported by this database/driver. Defaults to
+	 *         {@link Collections#emptySet()}.
+	 */
+	default Collection<? extends Class<?>> getSimpleTypes() {
+		return Collections.emptySet();
+	}
+
+	/**
+	 * Return the {@link SimpleTypeHolder} for this dialect.
+	 *
+	 * @return the {@link SimpleTypeHolder} for this dialect.
+	 * @see #getSimpleTypes()
+	 */
+	default SimpleTypeHolder getSimpleTypeHolder() {
+		return new SimpleTypeHolder(new HashSet<>(getSimpleTypes()), true);
+	}
+
 	/**
 	 * Return the {@link LimitClause} used by this dialect.
 	 *
 	 * @return the {@link LimitClause} used by this dialect.
 	 */
 	LimitClause limit();
+
+	/**
+	 * Returns {@literal true} whether this dialect supports array-typed column. Collection-typed columns can map their
+	 * content to native array types.
+	 *
+	 * @return {@literal true} whether this dialect supports array-typed columns.
+	 */
+	default boolean supportsArrayColumns() {
+		return false;
+	}
 }

src/main/java/org/springframework/data/r2dbc/dialect/PostgresDialect.java

@@ -1,5 +1,15 @@
 package org.springframework.data.r2dbc.dialect;
 
+import java.net.InetAddress;
+import java.net.URI;
+import java.net.URL;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+
 /**
  * An SQL dialect for Postgres.
  *
@@ -7,6 +17,9 @@ package org.springframework.data.r2dbc.dialect;
  */
 public class PostgresDialect implements Dialect {
 
+	private static final Set<Class<?>> SIMPLE_TYPES = new HashSet<>(
+			Arrays.asList(List.class, Collection.class, String[].class, UUID.class, URL.class, URI.class, InetAddress.class));
+
 	/**
 	 * Singleton instance.
 	 */
@@ -62,6 +75,15 @@ public class PostgresDialect implements Dialect {
 		return "RETURNING *";
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.r2dbc.dialect.Dialect#getSimpleTypesKeys()
+	 */
+	@Override
+	public Collection<? extends Class<?>> getSimpleTypes() {
+		return SIMPLE_TYPES;
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.r2dbc.dialect.Dialect#limit()
@@ -70,4 +92,13 @@ public class PostgresDialect implements Dialect {
 	public LimitClause limit() {
 		return LIMIT_CLAUSE;
 	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.r2dbc.dialect.Dialect#supportsArrayColumns()
+	 */
+	@Override
+	public boolean supportsArrayColumns() {
+		return true;
+	}
 }

src/main/java/org/springframework/data/r2dbc/dialect/SqlServerDialect.java

@@ -1,5 +1,11 @@
 package org.springframework.data.r2dbc.dialect;
 
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.UUID;
+
 /**
  * An SQL dialect for Microsoft SQL Server.
  *
@@ -7,6 +13,8 @@ package org.springframework.data.r2dbc.dialect;
  */
 public class SqlServerDialect implements Dialect {
 
+	private static final Set<Class<?>> SIMPLE_TYPES = new HashSet<>(Collections.singletonList(UUID.class));
+
 	/**
 	 * Singleton instance.
 	 */
@@ -63,6 +71,15 @@ public class SqlServerDialect implements Dialect {
 		return "select SCOPE_IDENTITY() AS GENERATED_KEYS";
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.r2dbc.dialect.Dialect#getSimpleTypesKeys()
+	 */
+	@Override
+	public Collection<? extends Class<?>> getSimpleTypes() {
+		return SIMPLE_TYPES;
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.r2dbc.dialect.Dialect#limit()

src/main/java/org/springframework/data/r2dbc/function/DefaultReactiveDataAccessStrategy.java

@@ -19,6 +19,7 @@ import io.r2dbc.spi.Row;
 import io.r2dbc.spi.RowMetadata;
 import io.r2dbc.spi.Statement;
 
+import java.lang.reflect.Array;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -29,22 +30,28 @@ import java.util.Set;
 import java.util.function.BiFunction;
 import java.util.function.Function;
 
+import org.springframework.dao.InvalidDataAccessApiUsageException;
+import org.springframework.dao.InvalidDataAccessResourceUsageException;
+import org.springframework.data.convert.CustomConversions.StoreConversions;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.domain.Sort.Order;
 import org.springframework.data.mapping.PersistentPropertyAccessor;
+import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.r2dbc.dialect.BindMarker;
 import org.springframework.data.r2dbc.dialect.BindMarkers;
 import org.springframework.data.r2dbc.dialect.Dialect;
 import org.springframework.data.r2dbc.dialect.LimitClause;
 import org.springframework.data.r2dbc.dialect.LimitClause.Position;
 import org.springframework.data.r2dbc.function.convert.EntityRowMapper;
+import org.springframework.data.r2dbc.function.convert.R2dbcCustomConversions;
 import org.springframework.data.r2dbc.function.convert.SettableValue;
 import org.springframework.data.relational.core.conversion.BasicRelationalConverter;
 import org.springframework.data.relational.core.conversion.RelationalConverter;
 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.data.util.TypeInformation;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
@@ -57,8 +64,9 @@ import org.springframework.util.StringUtils;
  */
 public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStrategy {
 
-	private final RelationalConverter relationalConverter;
 	private final Dialect dialect;
+	private final RelationalConverter relationalConverter;
+	private final MappingContext<RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty> mappingContext;
 
 	/**
 	 * Creates a new {@link DefaultReactiveDataAccessStrategy} given {@link Dialect}.
@@ -66,7 +74,28 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 	 * @param dialect the {@link Dialect} to use.
 	 */
 	public DefaultReactiveDataAccessStrategy(Dialect dialect) {
-		this(dialect, new BasicRelationalConverter(new RelationalMappingContext()));
+		this(dialect, createConverter(dialect));
+	}
+
+	private static BasicRelationalConverter createConverter(Dialect dialect) {
+
+		Assert.notNull(dialect, "Dialect must not be null");
+
+		R2dbcCustomConversions customConversions = new R2dbcCustomConversions(
+				StoreConversions.of(dialect.getSimpleTypeHolder()), Collections.emptyList());
+
+		RelationalMappingContext context = new RelationalMappingContext();
+		context.setSimpleTypeHolder(customConversions.getSimpleTypeHolder());
+
+		return new BasicRelationalConverter(context, customConversions);
+	}
+
+	public RelationalConverter getRelationalConverter() {
+		return relationalConverter;
+	}
+
+	public MappingContext<RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty> getMappingContext() {
+		return mappingContext;
 	}
 
 	/**
@@ -75,12 +104,15 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 	 * @param dialect the {@link Dialect} to use.
 	 * @param converter must not be {@literal null}.
 	 */
+	@SuppressWarnings("unchecked")
 	public DefaultReactiveDataAccessStrategy(Dialect dialect, RelationalConverter converter) {
 
 		Assert.notNull(dialect, "Dialect must not be null");
 		Assert.notNull(converter, "RelationalConverter must not be null");
 
 		this.relationalConverter = converter;
+		this.mappingContext = (MappingContext<RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty>) relationalConverter
+				.getMappingContext();
 		this.dialect = dialect;
 	}
 
@@ -121,7 +153,7 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 
 		for (RelationalPersistentProperty property : entity) {
 
-			Object value = propertyAccessor.getProperty(property);
+			Object value = getWriteValue(propertyAccessor, property);
 
 			if (value == null) {
 				continue;
@@ -133,6 +165,31 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 		return values;
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#getColumnsToUpdate(java.lang.Object)
+	 */
+	public Map<String, SettableValue> getColumnsToUpdate(Object object) {
+
+		Assert.notNull(object, "Entity object must not be null!");
+
+		Class<?> userClass = ClassUtils.getUserClass(object);
+		RelationalPersistentEntity<?> entity = getRequiredPersistentEntity(userClass);
+
+		Map<String, SettableValue> update = new LinkedHashMap<>();
+
+		PersistentPropertyAccessor propertyAccessor = entity.getPropertyAccessor(object);
+
+		for (RelationalPersistentProperty property : entity) {
+
+			Object writeValue = getWriteValue(propertyAccessor, property);
+
+			update.put(property.getColumnName(), new SettableValue(property.getColumnName(), writeValue, property.getType()));
+		}
+
+		return update;
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy#getMappedSort(java.lang.Class, org.springframework.data.domain.Sort)
@@ -181,12 +238,40 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 	}
 
 	private RelationalPersistentEntity<?> getRequiredPersistentEntity(Class<?> typeToRead) {
-		return relationalConverter.getMappingContext().getRequiredPersistentEntity(typeToRead);
+		return mappingContext.getRequiredPersistentEntity(typeToRead);
 	}
 
 	@Nullable
 	private RelationalPersistentEntity<?> getPersistentEntity(Class<?> typeToRead) {
-		return relationalConverter.getMappingContext().getPersistentEntity(typeToRead);
+		return mappingContext.getPersistentEntity(typeToRead);
+	}
+
+	private Object getWriteValue(PersistentPropertyAccessor propertyAccessor, RelationalPersistentProperty property) {
+
+		TypeInformation<?> type = property.getTypeInformation();
+		Object value = relationalConverter.writeValue(propertyAccessor.getProperty(property), type);
+
+		if (type.isCollectionLike()) {
+
+			RelationalPersistentEntity<?> nestedEntity = mappingContext
+					.getPersistentEntity(type.getRequiredActualType().getType());
+
+			if (nestedEntity != null) {
+				throw new InvalidDataAccessApiUsageException("Nested entities are not supported");
+			}
+
+			if (!dialect.supportsArrayColumns()) {
+				throw new InvalidDataAccessResourceUsageException(
+						"Dialect " + dialect.getClass().getName() + " does not support array columns");
+			}
+
+			if (!property.isArray()) {
+				Object zeroLengthArray = Array.newInstance(property.getActualType(), 0);
+				return relationalConverter.getConversionService().convert(value, zeroLengthArray.getClass());
+			}
+		}
+
+		return value;
 	}
 
 	/*

src/main/java/org/springframework/data/r2dbc/function/ReactiveDataAccessStrategy.java

@@ -20,6 +20,7 @@ import io.r2dbc.spi.RowMetadata;
 import io.r2dbc.spi.Statement;
 
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.function.BiFunction;
 
@@ -49,6 +50,14 @@ public interface ReactiveDataAccessStrategy {
 	 */
 	List<SettableValue> getValuesToInsert(Object object);
 
+	/**
+	 * Returns a {@link Map} that maps column names to a {@link SettableValue} value.
+	 *
+	 * @param object must not be {@literal null}.
+	 * @return
+	 */
+	Map<String, SettableValue> getColumnsToUpdate(Object object);
+
 	/**
 	 * Map the {@link Sort} object to apply field name mapping using {@link Class the type to read}.
 	 *

src/main/java/org/springframework/data/r2dbc/function/convert/MappingR2dbcConverter.java

@@ -21,7 +21,6 @@ import io.r2dbc.spi.RowMetadata;
 
 import java.util.LinkedHashMap;
 import java.util.Map;
-import java.util.Optional;
 import java.util.function.BiFunction;
 
 import org.springframework.core.convert.ConversionService;
@@ -65,32 +64,6 @@ public class MappingR2dbcConverter {
 		this.relationalConverter = converter;
 	}
 
-	/**
-	 * Returns a {@link Map} that maps column names to an {@link Optional} value. Used {@link Optional#empty()} if the
-	 * underlying property is {@literal null}.
-	 *
-	 * @param object must not be {@literal null}.
-	 * @return
-	 */
-	public Map<String, SettableValue> getColumnsToUpdate(Object object) {
-
-		Assert.notNull(object, "Entity object must not be null!");
-
-		Class<?> userClass = ClassUtils.getUserClass(object);
-		RelationalPersistentEntity<?> entity = getMappingContext().getRequiredPersistentEntity(userClass);
-
-		Map<String, SettableValue> update = new LinkedHashMap<>();
-
-		PersistentPropertyAccessor propertyAccessor = entity.getPropertyAccessor(object);
-
-		for (RelationalPersistentProperty property : entity) {
-			update.put(property.getColumnName(),
-					new SettableValue(property.getColumnName(), propertyAccessor.getProperty(property), property.getType()));
-		}
-
-		return update;
-	}
-
 	/**
 	 * Returns a {@link java.util.function.Function} that populates the id property of the {@code object} from a
 	 * {@link Row}.

src/main/java/org/springframework/data/r2dbc/function/convert/R2dbcCustomConversions.java

@@ -0,0 +1,26 @@
+package org.springframework.data.r2dbc.function.convert;
+
+import java.util.Collection;
+
+import org.springframework.data.convert.CustomConversions;
+
+/**
+ * Value object to capture custom conversion. {@link R2dbcCustomConversions} also act as factory for
+ * {@link org.springframework.data.mapping.model.SimpleTypeHolder}
+ *
+ * @author Mark Paluch
+ * @see CustomConversions
+ * @see org.springframework.data.mapping.model.SimpleTypeHolder
+ */
+public class R2dbcCustomConversions extends CustomConversions {
+
+	/**
+	 * Creates a new {@link CustomConversions} instance registering the given converters.
+	 *
+	 * @param storeConversions must not be {@literal null}.
+	 * @param converters must not be {@literal null}.
+	 */
+	public R2dbcCustomConversions(StoreConversions storeConversions, Collection<?> converters) {
+		super(storeConversions, converters);
+	}
+}

src/main/java/org/springframework/data/r2dbc/repository/config/R2dbcRepositoriesRegistrar.java

@@ -24,7 +24,6 @@ import org.springframework.data.repository.config.RepositoryConfigurationExtensi
  * R2DBC-specific {@link org.springframework.context.annotation.ImportBeanDefinitionRegistrar}.
  *
  * @author Mark Paluch
- * @since 2.0
  */
 class R2dbcRepositoriesRegistrar extends RepositoryBeanDefinitionRegistrarSupport {
 

src/main/java/org/springframework/data/r2dbc/repository/support/SimpleR2dbcRepository.java

@@ -70,7 +70,7 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
 		}
 
 		Object id = entity.getRequiredId(objectToSave);
-		Map<String, SettableValue> columns = converter.getColumnsToUpdate(objectToSave);
+		Map<String, SettableValue> columns = accessStrategy.getColumnsToUpdate(objectToSave);
 		columns.remove(getIdColumnName()); // do not update the Id column.
 		String idColumnName = getIdColumnName();
 		BindIdOperation update = accessStrategy.updateById(entity.getTableName(), columns.keySet(), idColumnName);