#2 - Adapt to R2DBC API changes.

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

@@ -348,8 +348,9 @@ public interface DatabaseClient {
 		 * Specify a {@literal null} value to insert.
 		 *
 		 * @param field must not be {@literal null} or empty.
+		 * @param type must not be {@literal null}.
 		 */
-		GenericInsertSpec nullValue(String field);
+		GenericInsertSpec nullValue(String field, Class<?> type);
 	}
 
 	/**
@@ -417,8 +418,9 @@ public interface DatabaseClient {
 		 * Bind a {@literal null} value to a parameter identified by its {@code index}.
 		 *
 		 * @param index
+		 * @param type must not be {@literal null}.
 		 */
-		S bindNull(int index);
+		S bindNull(int index, Class<?> type);
 
 		/**
 		 * Bind a non-{@literal null} value to a parameter identified by its {@code name}.
@@ -432,8 +434,9 @@ public interface DatabaseClient {
 		 * Bind a {@literal null} value to a parameter identified by its {@code name}.
 		 *
 		 * @param name must not be {@literal null} or empty.
+		 * @param type must not be {@literal null}.
 		 */
-		S bindNull(String name);
+		S bindNull(String name, Class<?> type);
 
 		/**
 		 * Bind a bean according to Java {@link java.beans.BeanInfo Beans} using property names.

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

@@ -36,7 +36,6 @@ import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.BiFunction;
 import java.util.function.Function;
@@ -53,10 +52,10 @@ import org.springframework.data.domain.Sort;
 import org.springframework.data.domain.Sort.NullHandling;
 import org.springframework.data.domain.Sort.Order;
 import org.springframework.data.r2dbc.UncategorizedR2dbcException;
+import org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy.SettableValue;
 import org.springframework.data.r2dbc.function.connectionfactory.ConnectionProxy;
 import org.springframework.data.r2dbc.function.convert.ColumnMapRowMapper;
 import org.springframework.data.r2dbc.support.R2dbcExceptionTranslator;
-import org.springframework.data.util.Pair;
 import org.springframework.jdbc.core.SqlProvider;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -217,24 +216,24 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 		return (dae != null ? dae : new UncategorizedR2dbcException(task, sql, ex));
 	}
 
-	private static void doBind(Statement statement, Map<String, Optional<Object>> byName,
-			Map<Integer, Optional<Object>> byIndex) {
+	private static void doBind(Statement statement, Map<String, SettableValue> byName,
+			Map<Integer, SettableValue> byIndex) {
 
 		byIndex.forEach((i, o) -> {
 
-			if (o.isPresent()) {
-				o.ifPresent(v -> statement.bind(i, v));
+			if (o.getValue() != null) {
+				statement.bind(i, o.getValue());
 			} else {
-				statement.bindNull(i, 0); // TODO: What is type?
+				statement.bindNull(i, o.getType());
 			}
 		});
 
 		byName.forEach((name, o) -> {
 
-			if (o.isPresent()) {
-				o.ifPresent(v -> statement.bind(name, v));
+			if (o.getValue() != null) {
+				statement.bind(name, o.getValue());
 			} else {
-				statement.bindNull(name, 0); // TODO: What is type?
+				statement.bindNull(name, o.getType());
 			}
 		});
 
@@ -267,8 +266,8 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 	@RequiredArgsConstructor
 	private class GenericExecuteSpecSupport {
 
-		final Map<Integer, Optional<Object>> byIndex;
-		final Map<String, Optional<Object>> byName;
+		final Map<Integer, SettableValue> byIndex;
+		final Map<String, SettableValue> byName;
 		final Supplier<String> sqlSupplier;
 
 		GenericExecuteSpecSupport(Supplier<String> sqlSupplier) {
@@ -310,16 +309,16 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 
 		public GenericExecuteSpecSupport bind(int index, Object value) {
 
-			Map<Integer, Optional<Object>> byIndex = new LinkedHashMap<>(this.byIndex);
-			byIndex.put(index, Optional.of(value));
+			Map<Integer, SettableValue> byIndex = new LinkedHashMap<>(this.byIndex);
+			byIndex.put(index, new SettableValue(index, value, null));
 
 			return createInstance(byIndex, this.byName, this.sqlSupplier);
 		}
 
-		public GenericExecuteSpecSupport bindNull(int index) {
+		public GenericExecuteSpecSupport bindNull(int index, Class<?> type) {
 
-			Map<Integer, Optional<Object>> byIndex = new LinkedHashMap<>(this.byIndex);
-			byIndex.put(index, Optional.empty());
+			Map<Integer, SettableValue> byIndex = new LinkedHashMap<>(this.byIndex);
+			byIndex.put(index, new SettableValue(index, null, type));
 
 			return createInstance(byIndex, this.byName, this.sqlSupplier);
 		}
@@ -328,24 +327,24 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 
 			Assert.hasText(name, "Parameter name must not be null or empty!");
 
-			Map<String, Optional<Object>> byName = new LinkedHashMap<>(this.byName);
-			byName.put(name, Optional.of(value));
+			Map<String, SettableValue> byName = new LinkedHashMap<>(this.byName);
+			byName.put(name, new SettableValue(name, value, null));
 
 			return createInstance(this.byIndex, byName, this.sqlSupplier);
 		}
 
-		public GenericExecuteSpecSupport bindNull(String name) {
+		public GenericExecuteSpecSupport bindNull(String name, Class<?> type) {
 
 			Assert.hasText(name, "Parameter name must not be null or empty!");
 
-			Map<String, Optional<Object>> byName = new LinkedHashMap<>(this.byName);
-			byName.put(name, Optional.empty());
+			Map<String, SettableValue> byName = new LinkedHashMap<>(this.byName);
+			byName.put(name, new SettableValue(name, null, type));
 
 			return createInstance(this.byIndex, byName, this.sqlSupplier);
 		}
 
-		protected GenericExecuteSpecSupport createInstance(Map<Integer, Optional<Object>> byIndex,
-				Map<String, Optional<Object>> byName, Supplier<String> sqlSupplier) {
+		protected GenericExecuteSpecSupport createInstance(Map<Integer, SettableValue> byIndex,
+				Map<String, SettableValue> byName, Supplier<String> sqlSupplier) {
 			return new GenericExecuteSpecSupport(byIndex, byName, sqlSupplier);
 		}
 
@@ -362,7 +361,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 	 */
 	private class DefaultGenericExecuteSpec extends GenericExecuteSpecSupport implements GenericExecuteSpec {
 
-		DefaultGenericExecuteSpec(Map<Integer, Optional<Object>> byIndex, Map<String, Optional<Object>> byName,
+		DefaultGenericExecuteSpec(Map<Integer, SettableValue> byIndex, Map<String, SettableValue> byName,
 				Supplier<String> sqlSupplier) {
 			super(byIndex, byName, sqlSupplier);
 		}
@@ -395,8 +394,8 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 		}
 
 		@Override
-		public DefaultGenericExecuteSpec bindNull(int index) {
-			return (DefaultGenericExecuteSpec) super.bindNull(index);
+		public DefaultGenericExecuteSpec bindNull(int index, Class<?> type) {
+			return (DefaultGenericExecuteSpec) super.bindNull(index, type);
 		}
 
 		@Override
@@ -405,8 +404,8 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 		}
 
 		@Override
-		public DefaultGenericExecuteSpec bindNull(String name) {
-			return (DefaultGenericExecuteSpec) super.bindNull(name);
+		public DefaultGenericExecuteSpec bindNull(String name, Class<?> type) {
+			return (DefaultGenericExecuteSpec) super.bindNull(name, type);
 		}
 
 		@Override
@@ -415,8 +414,8 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 		}
 
 		@Override
-		protected GenericExecuteSpecSupport createInstance(Map<Integer, Optional<Object>> byIndex,
-				Map<String, Optional<Object>> byName, Supplier<String> sqlSupplier) {
+		protected GenericExecuteSpecSupport createInstance(Map<Integer, SettableValue> byIndex,
+				Map<String, SettableValue> byName, Supplier<String> sqlSupplier) {
 			return new DefaultGenericExecuteSpec(byIndex, byName, sqlSupplier);
 		}
 	}
@@ -430,7 +429,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 		private final Class<T> typeToRead;
 		private final BiFunction<Row, RowMetadata, T> mappingFunction;
 
-		DefaultTypedGenericExecuteSpec(Map<Integer, Optional<Object>> byIndex, Map<String, Optional<Object>> byName,
+		DefaultTypedGenericExecuteSpec(Map<Integer, SettableValue> byIndex, Map<String, SettableValue> byName,
 				Supplier<String> sqlSupplier, Class<T> typeToRead) {
 
 			super(byIndex, byName, sqlSupplier);
@@ -463,8 +462,8 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 		}
 
 		@Override
-		public DefaultTypedGenericExecuteSpec<T> bindNull(int index) {
-			return (DefaultTypedGenericExecuteSpec<T>) super.bindNull(index);
+		public DefaultTypedGenericExecuteSpec<T> bindNull(int index, Class<?> type) {
+			return (DefaultTypedGenericExecuteSpec<T>) super.bindNull(index, type);
 		}
 
 		@Override
@@ -473,8 +472,8 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 		}
 
 		@Override
-		public DefaultTypedGenericExecuteSpec<T> bindNull(String name) {
-			return (DefaultTypedGenericExecuteSpec<T>) super.bindNull(name);
+		public DefaultTypedGenericExecuteSpec<T> bindNull(String name, Class<?> type) {
+			return (DefaultTypedGenericExecuteSpec<T>) super.bindNull(name, type);
 		}
 
 		@Override
@@ -483,8 +482,8 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 		}
 
 		@Override
-		protected DefaultTypedGenericExecuteSpec<T> createInstance(Map<Integer, Optional<Object>> byIndex,
-				Map<String, Optional<Object>> byName, Supplier<String> sqlSupplier) {
+		protected DefaultTypedGenericExecuteSpec<T> createInstance(Map<Integer, SettableValue> byIndex,
+				Map<String, SettableValue> byName, Supplier<String> sqlSupplier) {
 			return new DefaultTypedGenericExecuteSpec<>(byIndex, byName, sqlSupplier, typeToRead);
 		}
 	}
@@ -800,26 +799,26 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 	class DefaultGenericInsertSpec implements GenericInsertSpec {
 
 		private final String table;
-		private final Map<String, Optional<Object>> byName;
+		private final Map<String, SettableValue> byName;
 
 		@Override
 		public GenericInsertSpec value(String field, Object value) {
 
 			Assert.notNull(field, "Field must not be null!");
 
-			Map<String, Optional<Object>> byName = new LinkedHashMap<>(this.byName);
-			byName.put(field, Optional.of(value));
+			Map<String, SettableValue> byName = new LinkedHashMap<>(this.byName);
+			byName.put(field, new SettableValue(field, value, null));
 
 			return new DefaultGenericInsertSpec(this.table, byName);
 		}
 
 		@Override
-		public GenericInsertSpec nullValue(String field) {
+		public GenericInsertSpec nullValue(String field, Class<?> type) {
 
 			Assert.notNull(field, "Field must not be null!");
 
-			Map<String, Optional<Object>> byName = new LinkedHashMap<>(this.byName);
-			byName.put(field, Optional.empty());
+			Map<String, SettableValue> byName = new LinkedHashMap<>(this.byName);
+			byName.put(field, new SettableValue(field, null, type));
 
 			return new DefaultGenericInsertSpec(this.table, byName);
 		}
@@ -878,12 +877,12 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 
 			AtomicInteger index = new AtomicInteger();
 
-			for (Optional<Object> o : byName.values()) {
+			for (SettableValue value : byName.values()) {
 
-				if (o.isPresent()) {
-					o.ifPresent(v -> statement.bind(index.getAndIncrement(), v));
+				if (value.getValue() != null) {
+					statement.bind(index.getAndIncrement(), value.getValue());
 				} else {
-					statement.bindNull("$" + (index.getAndIncrement() + 1), 0); // TODO: What is type?
+					statement.bindNull("$" + (index.getAndIncrement() + 1), value.getType());
 				}
 			}
 		}
@@ -944,8 +943,9 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 
 			StringBuilder builder = new StringBuilder();
 
-			List<Pair<String, Object>> insertValues = dataAccessStrategy.getInsert(toInsert);
-			String fieldNames = insertValues.stream().map(Pair::getFirst).collect(Collectors.joining(","));
+			List<SettableValue> insertValues = dataAccessStrategy.getInsert(toInsert);
+			String fieldNames = insertValues.stream().map(SettableValue::getIdentifier).map(Object::toString)
+					.collect(Collectors.joining(","));
 			String placeholders = IntStream.range(0, insertValues.size()).mapToObj(i -> "$" + (i + 1))
 					.collect(Collectors.joining(","));
 
@@ -964,12 +964,12 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 
 				AtomicInteger index = new AtomicInteger();
 
-				for (Pair<String, Object> pair : insertValues) {
+				for (SettableValue settable : insertValues) {
 
-					if (pair.getSecond() != null) { // TODO: Better type to transport null values.
-						statement.bind(index.getAndIncrement(), pair.getSecond());
+					if (settable.getValue() != null) {
+						statement.bind(index.getAndIncrement(), settable.getValue());
 					} else {
-						statement.bindNull("$" + (index.getAndIncrement() + 1), 0); // TODO: What is type?
+						statement.bindNull("$" + (index.getAndIncrement() + 1), settable.getType());
 					}
 				}
 

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

@@ -24,16 +24,17 @@ import java.util.List;
 import java.util.function.BiFunction;
 import java.util.stream.Collectors;
 
-import org.springframework.data.convert.EntityInstantiators;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.domain.Sort.Order;
 import org.springframework.data.mapping.PersistentPropertyAccessor;
 import org.springframework.data.r2dbc.function.convert.EntityRowMapper;
+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.Pair;
 import org.springframework.data.util.StreamUtils;
+import org.springframework.lang.Nullable;
 import org.springframework.util.ClassUtils;
 
 /**
@@ -41,22 +42,20 @@ import org.springframework.util.ClassUtils;
  */
 public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStrategy {
 
-	private final RelationalMappingContext mappingContext;
-	private final EntityInstantiators instantiators;
+	private final RelationalConverter relationalConverter;
 
 	public DefaultReactiveDataAccessStrategy() {
-		this(new RelationalMappingContext(), new EntityInstantiators());
+		this(new BasicRelationalConverter(new RelationalMappingContext()));
 	}
 
-	public DefaultReactiveDataAccessStrategy(RelationalMappingContext mappingContext, EntityInstantiators instantiators) {
-		this.mappingContext = mappingContext;
-		this.instantiators = instantiators;
+	public DefaultReactiveDataAccessStrategy(RelationalConverter converter) {
+		this.relationalConverter = converter;
 	}
 
 	@Override
 	public List<String> getAllFields(Class<?> typeToRead) {
 
-		RelationalPersistentEntity<?> persistentEntity = mappingContext.getPersistentEntity(typeToRead);
+		RelationalPersistentEntity<?> persistentEntity = getPersistentEntity(typeToRead);
 
 		if (persistentEntity == null) {
 			return Collections.singletonList("*");
@@ -68,14 +67,14 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 	}
 
 	@Override
-	public List<Pair<String, Object>> getInsert(Object object) {
+	public List<SettableValue> getInsert(Object object) {
 
 		Class<?> userClass = ClassUtils.getUserClass(object);
 
-		RelationalPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(userClass);
+		RelationalPersistentEntity<?> entity = getRequiredPersistentEntity(userClass);
 		PersistentPropertyAccessor propertyAccessor = entity.getPropertyAccessor(object);
 
-		List<Pair<String, Object>> values = new ArrayList<>();
+		List<SettableValue> values = new ArrayList<>();
 
 		for (RelationalPersistentProperty property : entity) {
 
@@ -85,7 +84,7 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 				continue;
 			}
 
-			values.add(Pair.of(property.getColumnName(), value));
+			values.add(new SettableValue(property.getColumnName(), value, property.getType()));
 		}
 
 		return values;
@@ -94,7 +93,7 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 	@Override
 	public Sort getMappedSort(Class<?> typeToRead, Sort sort) {
 
-		RelationalPersistentEntity<?> entity = mappingContext.getPersistentEntity(typeToRead);
+		RelationalPersistentEntity<?> entity = getPersistentEntity(typeToRead);
 		if (entity == null) {
 			return sort;
 		}
@@ -117,12 +116,21 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 
 	@Override
 	public <T> BiFunction<Row, RowMetadata, T> getRowMapper(Class<T> typeToRead) {
-		return new EntityRowMapper<T>((RelationalPersistentEntity) mappingContext.getRequiredPersistentEntity(typeToRead),
-				instantiators, mappingContext);
+		return new EntityRowMapper<T>((RelationalPersistentEntity) getRequiredPersistentEntity(typeToRead),
+				relationalConverter);
 	}
 
 	@Override
 	public String getTableName(Class<?> type) {
-		return mappingContext.getRequiredPersistentEntity(type).getTableName();
+		return getRequiredPersistentEntity(type).getTableName();
+	}
+
+	private RelationalPersistentEntity<?> getRequiredPersistentEntity(Class<?> typeToRead) {
+		return relationalConverter.getMappingContext().getRequiredPersistentEntity(typeToRead);
+	}
+
+	@Nullable
+	private RelationalPersistentEntity<?> getPersistentEntity(Class<?> typeToRead) {
+		return relationalConverter.getMappingContext().getPersistentEntity(typeToRead);
 	}
 }

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

@@ -22,7 +22,7 @@ import java.util.List;
 import java.util.function.BiFunction;
 
 import org.springframework.data.domain.Sort;
-import org.springframework.data.util.Pair;
+import org.springframework.lang.Nullable;
 
 /**
  * @author Mark Paluch
@@ -31,7 +31,7 @@ public interface ReactiveDataAccessStrategy {
 
 	List<String> getAllFields(Class<?> typeToRead);
 
-	List<Pair<String, Object>> getInsert(Object object);
+	List<SettableValue> getInsert(Object object);
 
 	Sort getMappedSort(Class<?> typeToRead, Sort sort);
 
@@ -39,4 +39,55 @@ public interface ReactiveDataAccessStrategy {
 	<T> BiFunction<Row, RowMetadata, T> getRowMapper(Class<T> typeToRead);
 
 	String getTableName(Class<?> type);
+
+	/**
+	 * A database value that can be set in a statement.
+	 */
+	class SettableValue {
+
+		private final Object identifier;
+		private final @Nullable Object value;
+		private final Class<?> type;
+
+		/**
+		 * Create a {@link SettableValue} using an integer index.
+		 *
+		 * @param index
+		 * @param value
+		 * @param type
+		 */
+		public SettableValue(int index, @Nullable Object value, Class<?> type) {
+
+			this.identifier = index;
+			this.value = value;
+			this.type = type;
+		}
+
+		/**
+		 * Create a {@link SettableValue} using a {@link String} identifier.
+		 *
+		 * @param identifier
+		 * @param value
+		 * @param type
+		 */
+		public SettableValue(String identifier, @Nullable Object value, Class<?> type) {
+
+			this.identifier = identifier;
+			this.value = value;
+			this.type = type;
+		}
+
+		public Object getIdentifier() {
+			return identifier;
+		}
+
+		@Nullable
+		public Object getValue() {
+			return value;
+		}
+
+		public Class<?> getType() {
+			return type;
+		}
+	}
 }

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

@@ -89,7 +89,7 @@ public class ColumnMapRowMapper implements BiFunction<Row, RowMetadata, Map<Stri
 	/**
 	 * Retrieve a R2DBC object value for the specified column.
 	 * <p>
-	 * The default implementation uses the {@link Row#get(Object, Class)} method.
+	 * The default implementation uses the {@link Row#get(Object)} method.
 	 *
 	 * @param row is the {@link Row} holding the data.
 	 * @param index is the column index.
@@ -97,6 +97,6 @@ public class ColumnMapRowMapper implements BiFunction<Row, RowMetadata, Map<Stri
 	 */
 	@Nullable
 	protected Object getColumnValue(Row row, int index) {
-		return row.get(index, Object.class);
+		return row.get(index);
 	}
 }

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

@@ -24,18 +24,15 @@ import java.sql.ResultSet;
 import java.util.function.BiFunction;
 
 import org.springframework.core.convert.ConversionService;
-import org.springframework.data.convert.EntityInstantiators;
 import org.springframework.data.mapping.MappingException;
 import org.springframework.data.mapping.PersistentProperty;
 import org.springframework.data.mapping.PersistentPropertyAccessor;
 import org.springframework.data.mapping.PreferredConstructor.Parameter;
-import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.mapping.model.ConvertingPropertyAccessor;
 import org.springframework.data.mapping.model.ParameterValueProvider;
-import org.springframework.data.relational.core.mapping.RelationalMappingContext;
+import org.springframework.data.relational.core.conversion.RelationalConverter;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
-import org.springframework.util.ClassUtils;
 
 /**
  * Maps a {@link io.r2dbc.spi.Row} to an entity of type {@code T}, including entities referenced.
@@ -46,17 +43,12 @@ import org.springframework.util.ClassUtils;
 public class EntityRowMapper<T> implements BiFunction<Row, RowMetadata, T> {
 
 	private final RelationalPersistentEntity<T> entity;
-	private final EntityInstantiators entityInstantiators;
-	private final ConversionService conversions;
-	private final MappingContext<RelationalPersistentEntity<?>, RelationalPersistentProperty> context;
+	private final RelationalConverter converter;
 
-	public EntityRowMapper(RelationalPersistentEntity<T> entity, EntityInstantiators entityInstantiators,
-			RelationalMappingContext context) {
+	public EntityRowMapper(RelationalPersistentEntity<T> entity, RelationalConverter converter) {
 
 		this.entity = entity;
-		this.entityInstantiators = entityInstantiators;
-		this.conversions = context.getConversions();
-		this.context = context;
+		this.converter = converter;
 	}
 
 	@Override
@@ -65,7 +57,7 @@ public class EntityRowMapper<T> implements BiFunction<Row, RowMetadata, T> {
 		T result = createInstance(row, "", entity);
 
 		ConvertingPropertyAccessor propertyAccessor = new ConvertingPropertyAccessor(entity.getPropertyAccessor(result),
-				conversions);
+				converter.getConversionService());
 
 		for (RelationalPersistentProperty property : entity) {
 
@@ -103,23 +95,19 @@ public class EntityRowMapper<T> implements BiFunction<Row, RowMetadata, T> {
 				return readEntityFrom(row, property);
 			}
 
-			return row.get(prefix + property.getColumnName(), getType(property));
+			return row.get(prefix + property.getColumnName());
 
 		} catch (Exception o_O) {
 			throw new MappingException(String.format("Could not read property %s from result set!", property), o_O);
 		}
 	}
 
-	private static Class<?> getType(RelationalPersistentProperty property) {
-		return ClassUtils.resolvePrimitiveIfNecessary(property.getActualType());
-	}
-
 	private <S> S readEntityFrom(Row row, PersistentProperty<?> property) {
 
 		String prefix = property.getName() + "_";
 
 		@SuppressWarnings("unchecked")
-		RelationalPersistentEntity<S> entity = (RelationalPersistentEntity<S>) context
+		RelationalPersistentEntity<S> entity = (RelationalPersistentEntity<S>) converter.getMappingContext()
 				.getRequiredPersistentEntity(property.getActualType());
 
 		if (readFrom(row, entity.getRequiredIdProperty(), prefix) == null) {
@@ -129,7 +117,8 @@ public class EntityRowMapper<T> implements BiFunction<Row, RowMetadata, T> {
 		S instance = createInstance(row, prefix, entity);
 
 		PersistentPropertyAccessor accessor = entity.getPropertyAccessor(instance);
-		ConvertingPropertyAccessor propertyAccessor = new ConvertingPropertyAccessor(accessor, conversions);
+		ConvertingPropertyAccessor propertyAccessor = new ConvertingPropertyAccessor(accessor,
+				converter.getConversionService());
 
 		for (RelationalPersistentProperty p : entity) {
 			if (!entity.isConstructorArgument(property)) {
@@ -142,8 +131,10 @@ public class EntityRowMapper<T> implements BiFunction<Row, RowMetadata, T> {
 
 	private <S> S createInstance(Row row, String prefix, RelationalPersistentEntity<S> entity) {
 
-		return entityInstantiators.getInstantiatorFor(entity).createInstance(entity,
-				new RowParameterValueProvider(row, entity, conversions, prefix));
+		RowParameterValueProvider rowParameterValueProvider = new RowParameterValueProvider(row, entity,
+				converter.getConversionService(), prefix);
+
+		return converter.createInstance(entity, rowParameterValueProvider::getParameterValue);
 	}
 
 	@RequiredArgsConstructor
@@ -164,7 +155,7 @@ public class EntityRowMapper<T> implements BiFunction<Row, RowMetadata, T> {
 			String column = prefix + entity.getRequiredPersistentProperty(parameter.getName()).getColumnName();
 
 			try {
-				return conversionService.convert(resultSet.get(column, parameter.getType().getType()),
+				return conversionService.convert(resultSet.get(column),
 						parameter.getType().getType());
 			} catch (Exception o_O) {
 				throw new MappingException(String.format("Couldn't read column %s from Row.", column), o_O);

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

@@ -23,8 +23,11 @@ import java.util.Map;
 import java.util.Optional;
 import java.util.function.BiFunction;
 
+import org.springframework.core.convert.ConversionService;
 import org.springframework.data.mapping.PersistentPropertyAccessor;
 import org.springframework.data.mapping.context.MappingContext;
+import org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy.SettableValue;
+import org.springframework.data.relational.core.conversion.RelationalConverter;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
 import org.springframework.util.Assert;
@@ -37,11 +40,10 @@ import org.springframework.util.ClassUtils;
  */
 public class MappingR2dbcConverter {
 
-	private final MappingContext<? extends RelationalPersistentEntity<?>, RelationalPersistentProperty> mappingContext;
+	private final RelationalConverter relationalConverter;
 
-	public MappingR2dbcConverter(
-			MappingContext<? extends RelationalPersistentEntity<?>, RelationalPersistentProperty> mappingContext) {
-		this.mappingContext = mappingContext;
+	public MappingR2dbcConverter(RelationalConverter converter) {
+		this.relationalConverter = converter;
 	}
 
 	/**
@@ -51,19 +53,20 @@ public class MappingR2dbcConverter {
 	 * @param object must not be {@literal null}.
 	 * @return
 	 */
-	public Map<String, Optional<Object>> getFieldsToUpdate(Object object) {
+	public Map<String, SettableValue> getFieldsToUpdate(Object object) {
 
 		Assert.notNull(object, "Entity object must not be null!");
 
 		Class<?> userClass = ClassUtils.getUserClass(object);
-		RelationalPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(userClass);
+		RelationalPersistentEntity<?> entity = getMappingContext().getRequiredPersistentEntity(userClass);
 
-		Map<String, Optional<Object>> update = new LinkedHashMap<>();
+		Map<String, SettableValue> update = new LinkedHashMap<>();
 
 		PersistentPropertyAccessor propertyAccessor = entity.getPropertyAccessor(object);
 
 		for (RelationalPersistentProperty property : entity) {
-			update.put(property.getColumnName(), Optional.ofNullable(propertyAccessor.getProperty(property)));
+			update.put(property.getColumnName(),
+					new SettableValue(property.getColumnName(), propertyAccessor.getProperty(property), property.getType()));
 		}
 
 		return update;
@@ -82,7 +85,7 @@ public class MappingR2dbcConverter {
 		Assert.notNull(object, "Entity object must not be null!");
 
 		Class<?> userClass = ClassUtils.getUserClass(object);
-		RelationalPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(userClass);
+		RelationalPersistentEntity<?> entity = getMappingContext().getRequiredPersistentEntity(userClass);
 
 		return (row, metadata) -> {
 
@@ -91,7 +94,11 @@ public class MappingR2dbcConverter {
 
 			if (propertyAccessor.getProperty(idProperty) == null) {
 
-				propertyAccessor.setProperty(idProperty, row.get(idProperty.getColumnName(), idProperty.getColumnType()));
+				ConversionService conversionService = relationalConverter.getConversionService();
+				Object value = row.get(idProperty.getColumnName());
+
+				propertyAccessor.setProperty(idProperty, conversionService.convert(value, idProperty.getType()));
+
 				return (T) propertyAccessor.getBean();
 			}
 
@@ -99,7 +106,8 @@ public class MappingR2dbcConverter {
 		};
 	}
 
-	public MappingContext<? extends RelationalPersistentEntity<?>, RelationalPersistentProperty> getMappingContext() {
-		return mappingContext;
+	public MappingContext<? extends RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty> getMappingContext() {
+		return relationalConverter.getMappingContext();
 	}
 }
+

src/main/java/org/springframework/data/r2dbc/repository/query/R2dbcQueryExecution.java

@@ -64,7 +64,7 @@ interface R2dbcQueryExecution {
 	final class ResultProcessingConverter implements Converter<Object, Object> {
 
 		private final @NonNull ResultProcessor processor;
-		private final @NonNull MappingContext<? extends RelationalPersistentEntity<?>, RelationalPersistentProperty> mappingContext;
+		private final @NonNull MappingContext<? extends RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty> mappingContext;
 		private final @NonNull EntityInstantiators instantiators;
 
 		/* (non-Javadoc)

src/main/java/org/springframework/data/r2dbc/repository/query/R2dbcQueryMethod.java

@@ -56,7 +56,7 @@ public class R2dbcQueryMethod extends QueryMethod {
 	private static final ClassTypeInformation<Slice> SLICE_TYPE = ClassTypeInformation.from(Slice.class);
 
 	private final Method method;
-	private final MappingContext<? extends RelationalPersistentEntity<?>, RelationalPersistentProperty> mappingContext;
+	private final MappingContext<? extends RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty> mappingContext;
 	private final Optional<Query> query;
 
 	private @Nullable RelationalEntityMetadata<?> metadata;
@@ -70,7 +70,7 @@ public class R2dbcQueryMethod extends QueryMethod {
 	 * @param mappingContext must not be {@literal null}.
 	 */
 	public R2dbcQueryMethod(Method method, RepositoryMetadata metadata, ProjectionFactory projectionFactory,
-			MappingContext<? extends RelationalPersistentEntity<?>, RelationalPersistentProperty> mappingContext) {
+			MappingContext<? extends RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty> mappingContext) {
 
 		super(method, metadata, projectionFactory);
 

src/main/java/org/springframework/data/r2dbc/repository/query/StringBasedR2dbcQuery.java

@@ -20,6 +20,8 @@ import org.springframework.data.r2dbc.function.DatabaseClient;
 import org.springframework.data.r2dbc.function.DatabaseClient.BindSpec;
 import org.springframework.data.r2dbc.function.convert.MappingR2dbcConverter;
 import org.springframework.data.relational.repository.query.RelationalParameterAccessor;
+import org.springframework.data.repository.query.Parameter;
+import org.springframework.data.repository.query.Parameters;
 import org.springframework.data.repository.query.QueryMethodEvaluationContextProvider;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.util.Assert;
@@ -88,12 +90,17 @@ public class StringBasedR2dbcQuery extends AbstractR2dbcQuery {
 				T bindSpecToUse = bindSpec;
 
 				// TODO: Encapsulate PostgreSQL-specific bindings
+
+				Parameters<?, ?> bindableParameters = accessor.getBindableParameters();
+
 				int index = 1;
 				for (Object value : accessor.getValues()) {
 
+					Parameter bindableParameter = bindableParameters.getBindableParameter(index - 1);
+
 					if (value == null) {
 						if (accessor.hasBindableNullValue()) {
-							bindSpecToUse = bindSpecToUse.bindNull("$" + (index++));
+							bindSpecToUse = bindSpecToUse.bindNull("$" + (index++), bindableParameter.getType());
 						}
 					} else {
 						bindSpecToUse = bindSpecToUse.bind("$" + (index++), value);

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

@@ -28,6 +28,7 @@ import org.springframework.data.r2dbc.function.convert.MappingR2dbcConverter;
 import org.springframework.data.r2dbc.repository.R2dbcRepository;
 import org.springframework.data.r2dbc.repository.query.R2dbcQueryMethod;
 import org.springframework.data.r2dbc.repository.query.StringBasedR2dbcQuery;
+import org.springframework.data.relational.core.conversion.BasicRelationalConverter;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
 import org.springframework.data.relational.repository.query.RelationalEntityInformation;
@@ -71,7 +72,7 @@ public class R2dbcRepositoryFactory extends ReactiveRepositoryFactorySupport {
 
 		this.databaseClient = databaseClient;
 		this.mappingContext = mappingContext;
-		this.converter = new MappingR2dbcConverter(mappingContext);
+		this.converter = new MappingR2dbcConverter(new BasicRelationalConverter(mappingContext));
 	}
 
 	/*

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

@@ -22,7 +22,6 @@ import reactor.core.publisher.Mono;
 
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
@@ -31,6 +30,7 @@ import org.springframework.data.r2dbc.function.DatabaseClient;
 import org.springframework.data.r2dbc.function.DatabaseClient.BindSpec;
 import org.springframework.data.r2dbc.function.DatabaseClient.GenericExecuteSpec;
 import org.springframework.data.r2dbc.function.FetchSpec;
+import org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy.SettableValue;
 import org.springframework.data.r2dbc.function.convert.MappingR2dbcConverter;
 import org.springframework.data.relational.repository.query.RelationalEntityInformation;
 import org.springframework.data.repository.reactive.ReactiveCrudRepository;
@@ -68,7 +68,7 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
 		// TODO: Extract in some kind of SQL generator
 		Object id = entity.getRequiredId(objectToSave);
 
-		Map<String, Optional<Object>> fields = converter.getFieldsToUpdate(objectToSave);
+		Map<String, SettableValue> fields = converter.getFieldsToUpdate(objectToSave);
 
 		String setClause = getSetClause(fields);
 
@@ -77,13 +77,13 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
 				.bind(0, id);
 
 		int index = 1;
-		for (Optional<Object> setValue : fields.values()) {
+		for (SettableValue setValue : fields.values()) {
 
-			Object value = setValue.orElse(null);
+			Object value = setValue.getValue();
 			if (value != null) {
 				exec = exec.bind(index++, value);
 			} else {
-				exec = exec.bindNull(index++);
+				exec = exec.bindNull(index++, setValue.getType());
 			}
 		}
 
@@ -93,7 +93,7 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
 				.thenReturn(objectToSave);
 	}
 
-	private static String getSetClause(Map<String, Optional<Object>> fields) {
+	private static String getSetClause(Map<String, ?> fields) {
 
 		StringBuilder setClause = new StringBuilder();
 

src/main/java/org/springframework/data/relational/repository/query/DtoInstantiatingConverter.java

@@ -49,7 +49,7 @@ public class DtoInstantiatingConverter implements Converter<Object, Object> {
 	 * @param instantiators must not be {@literal null}.
 	 */
 	public DtoInstantiatingConverter(Class<?> dtoType,
-			MappingContext<? extends RelationalPersistentEntity<?>, RelationalPersistentProperty> context,
+			MappingContext<? extends RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty> context,
 			EntityInstantiators instantiator) {
 
 		Assert.notNull(dtoType, "DTO type must not be null!");

src/main/java/org/springframework/data/relational/repository/query/RelationalParameterAccessor.java

@@ -16,6 +16,7 @@
 package org.springframework.data.relational.repository.query;
 
 import org.springframework.data.repository.query.ParameterAccessor;
+import org.springframework.data.repository.query.Parameters;
 
 /**
  * JDBC-specific {@link ParameterAccessor}.
@@ -28,4 +29,9 @@ public interface RelationalParameterAccessor extends ParameterAccessor {
 	 * Returns the raw parameter values of the underlying query method.
 	 */
 	Object[] getValues();
+
+	/**
+	 * @return the bindable parameters.
+	 */
+	Parameters<?, ?> getBindableParameters();
 }

src/main/java/org/springframework/data/relational/repository/query/RelationalParametersParameterAccessor.java

@@ -18,6 +18,7 @@ package org.springframework.data.relational.repository.query;
 import java.util.Arrays;
 import java.util.List;
 
+import org.springframework.data.repository.query.Parameters;
 import org.springframework.data.repository.query.ParametersParameterAccessor;
 import org.springframework.data.repository.query.QueryMethod;
 
@@ -50,4 +51,12 @@ public class RelationalParametersParameterAccessor extends ParametersParameterAc
 	public Object[] getValues() {
 		return values.toArray();
 	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.jdbc.repository.query.RelationalParameterAccessor#getBindableParameters()
+	 */
+	@Override
+	public Parameters<?, ?> getBindableParameters() {
+		return getParameters().getBindableParameters();
+	}
 }