diff --git a/src/main/java/org/springframework/data/r2dbc/convert/MappingR2dbcConverter.java b/src/main/java/org/springframework/data/r2dbc/convert/MappingR2dbcConverter.java
index 09250b70f..5ce6d2c97 100644
--- a/src/main/java/org/springframework/data/r2dbc/convert/MappingR2dbcConverter.java
+++ b/src/main/java/org/springframework/data/r2dbc/convert/MappingR2dbcConverter.java
@@ -162,15 +162,23 @@ public class MappingR2dbcConverter extends BasicRelationalConverter implements R
 
 		try {
 
+			Object value = null;
+			if (metadata == null || metadata.getColumnNames().contains(identifier)) {
+				value = row.get(identifier);
+			}
+
+			if (value != null && getConversions().hasCustomReadTarget(value.getClass(), property.getType())) {
+				return readValue(value, property.getTypeInformation());
+			}
+
 			if (property.isEntity()) {
 				return readEntityFrom(row, metadata, property);
 			}
 
-			if (metadata != null && !metadata.getColumnNames().contains(identifier)) {
+			if (value == null) {
 				return null;
 			}
 
-			Object value = row.get(identifier);
 			return readValue(value, property.getTypeInformation());
 
 		} catch (Exception o_O) {
@@ -270,7 +278,7 @@ public class MappingR2dbcConverter extends BasicRelationalConverter implements R
 	}
 
 	@SuppressWarnings("unchecked")
-	private <S> S readEntityFrom(Row row, RowMetadata metadata, PersistentProperty<?> property) {
+	private <S> S readEntityFrom(Row row, @Nullable RowMetadata metadata, PersistentProperty<?> property) {
 
 		String prefix = property.getName() + "_";
 
diff --git a/src/test/java/org/springframework/data/r2dbc/convert/PostgresMappingR2dbcConverterUnitTests.java b/src/test/java/org/springframework/data/r2dbc/convert/PostgresMappingR2dbcConverterUnitTests.java
index 928a325fc..cb25802b7 100644
--- a/src/test/java/org/springframework/data/r2dbc/convert/PostgresMappingR2dbcConverterUnitTests.java
+++ b/src/test/java/org/springframework/data/r2dbc/convert/PostgresMappingR2dbcConverterUnitTests.java
@@ -24,14 +24,21 @@ import io.r2dbc.spi.test.MockRowMetadata;
 import lombok.AllArgsConstructor;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.Set;
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import org.springframework.core.convert.TypeDescriptor;
+import org.springframework.core.convert.converter.ConditionalConverter;
+import org.springframework.core.convert.converter.GenericConverter;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.convert.CustomConversions;
+import org.springframework.data.convert.ReadingConverter;
+import org.springframework.data.convert.WritingConverter;
 import org.springframework.data.r2dbc.dialect.PostgresDialect;
 import org.springframework.data.r2dbc.mapping.OutboundRow;
 import org.springframework.data.r2dbc.mapping.R2dbcMappingContext;
@@ -44,20 +51,21 @@ import org.springframework.r2dbc.core.Parameter;
  *
  * @author Mark Paluch
  */
-public class PostgresMappingR2dbcConverterUnitTests {
+class PostgresMappingR2dbcConverterUnitTests {
 
-	RelationalMappingContext mappingContext = new R2dbcMappingContext();
-	MappingR2dbcConverter converter = new MappingR2dbcConverter(mappingContext);
+	private RelationalMappingContext mappingContext = new R2dbcMappingContext();
+	private MappingR2dbcConverter converter = new MappingR2dbcConverter(mappingContext);
 
 	@BeforeEach
-	public void before() {
+	void before() {
 
 		List<Object> converters = new ArrayList<>(PostgresDialect.INSTANCE.getConverters());
 		converters.addAll(R2dbcCustomConversions.STORE_CONVERTERS);
 		CustomConversions.StoreConversions storeConversions = CustomConversions.StoreConversions
 				.of(PostgresDialect.INSTANCE.getSimpleTypeHolder(), converters);
 
-		R2dbcCustomConversions customConversions = new R2dbcCustomConversions(storeConversions, Collections.emptyList());
+		R2dbcCustomConversions customConversions = new R2dbcCustomConversions(storeConversions,
+				Arrays.asList(JsonToJsonHolderConverter.INSTANCE, JsonHolderToJsonConverter.INSTANCE));
 
 		mappingContext.setSimpleTypeHolder(customConversions.getSimpleTypeHolder());
 
@@ -65,7 +73,7 @@ public class PostgresMappingR2dbcConverterUnitTests {
 	}
 
 	@Test // gh-318
-	public void shouldPassThruJson() {
+	void shouldPassThruJson() {
 
 		JsonPerson person = new JsonPerson(null, Json.of("{\"hello\":\"world\"}"));
 
@@ -76,7 +84,7 @@ public class PostgresMappingR2dbcConverterUnitTests {
 	}
 
 	@Test // gh-453
-	public void shouldConvertJsonToString() {
+	void shouldConvertJsonToString() {
 
 		MockRow row = MockRow.builder().identified("json_string", Object.class, Json.of("{\"hello\":\"world\"}")).build();
 
@@ -88,7 +96,7 @@ public class PostgresMappingR2dbcConverterUnitTests {
 	}
 
 	@Test // gh-453
-	public void shouldConvertJsonToByteArray() {
+	void shouldConvertJsonToByteArray() {
 
 		MockRow row = MockRow.builder().identified("json_bytes", Object.class, Json.of("{\"hello\":\"world\"}")).build();
 
@@ -99,6 +107,32 @@ public class PostgresMappingR2dbcConverterUnitTests {
 		assertThat(result.jsonBytes).isEqualTo("{\"hello\":\"world\"}".getBytes());
 	}
 
+	@Test // gh-585
+	void shouldApplyCustomReadingConverter() {
+
+		MockRow row = MockRow.builder().identified("holder", Object.class, Json.of("{\"hello\":\"world\"}")).build();
+
+		MockRowMetadata metadata = MockRowMetadata.builder()
+				.columnMetadata(MockColumnMetadata.builder().name("holder").build()).build();
+
+		WithJsonHolder result = converter.read(WithJsonHolder.class, row, metadata);
+		assertThat(result.holder).isNotNull();
+		assertThat(result.holder.json).isNotNull();
+	}
+
+	@Test // gh-585
+	void shouldApplyCustomWritingConverter() {
+
+		WithJsonHolder object = new WithJsonHolder(new JsonHolder(Json.of("{\"hello\":\"world\"}")));
+
+		OutboundRow row = new OutboundRow();
+		converter.write(object, row);
+
+		Parameter parameter = row.get(SqlIdentifier.unquoted("holder"));
+		assertThat(parameter).isNotNull();
+		assertThat(parameter.getValue()).isInstanceOf(Json.class);
+	}
+
 	@AllArgsConstructor
 	static class JsonPerson {
 
@@ -116,4 +150,60 @@ public class PostgresMappingR2dbcConverterUnitTests {
 
 		byte[] jsonBytes;
 	}
+
+	@AllArgsConstructor
+	static class WithJsonHolder {
+
+		JsonHolder holder;
+	}
+
+	@ReadingConverter
+	enum JsonToJsonHolderConverter implements GenericConverter, ConditionalConverter {
+
+		INSTANCE;
+
+		@Override
+		public boolean matches(TypeDescriptor sourceType, TypeDescriptor targetType) {
+			return Json.class.isAssignableFrom(sourceType.getType());
+		}
+
+		@Override
+		public Set<ConvertiblePair> getConvertibleTypes() {
+			return Collections.singleton(new GenericConverter.ConvertiblePair(Json.class, Object.class));
+		}
+
+		@Override
+		public Object convert(Object source, TypeDescriptor sourceType, TypeDescriptor targetType) {
+			return new JsonHolder((Json) source);
+		}
+	}
+
+	@WritingConverter
+	enum JsonHolderToJsonConverter implements GenericConverter, ConditionalConverter {
+
+		INSTANCE;
+
+		@Override
+		public boolean matches(TypeDescriptor sourceType, TypeDescriptor targetType) {
+			return JsonHolder.class.isAssignableFrom(sourceType.getType());
+		}
+
+		@Override
+		public Set<ConvertiblePair> getConvertibleTypes() {
+			return Collections.singleton(new GenericConverter.ConvertiblePair(JsonHolder.class, Json.class));
+		}
+
+		@Override
+		public Object convert(Object source, TypeDescriptor sourceType, TypeDescriptor targetType) {
+			return ((JsonHolder) source).json;
+		}
+	}
+
+	@AllArgsConstructor
+	private static class JsonHolder {
+
+		private final Json json;
+
+	}
+
 }