#30 - Address review feedback.

Introduce ArrayColumns type to encapsulate Dialect-specific array support.
Apply array conversion for properties that do not match the native array type.
Add integration tests for Postgres array columns.

Original pull request: #31.

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

@@ -0,0 +1,53 @@
+package org.springframework.data.r2dbc.dialect;
+
+/**
+ * Interface declaring methods that express how a dialect supports array-typed columns.
+ *
+ * @author Mark Paluch
+ */
+public interface ArrayColumns {
+
+	/**
+	 * Returns {@literal true} if the dialect supports array-typed columns.
+	 *
+	 * @return {@literal true} if the dialect supports array-typed columns.
+	 */
+	boolean isSupported();
+
+	/**
+	 * Translate the {@link Class user type} of an array into the dialect-specific type. This method considers only the
+	 * component type.
+	 *
+	 * @param userType component type of the array.
+	 * @return the dialect-supported array type.
+	 * @throws UnsupportedOperationException if array typed columns are not supported.
+	 * @throws IllegalArgumentException if the {@code userType} is not a supported array type.
+	 */
+	Class<?> getArrayType(Class<?> userType);
+
+	/**
+	 * Default {@link ArrayColumns} implementation for dialects that do not support array-typed columns.
+	 */
+	enum Unsupported implements ArrayColumns {
+
+		INSTANCE;
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.dialect.ArrayColumns#isSupported()
+		 */
+		@Override
+		public boolean isSupported() {
+			return false;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.dialect.ArrayColumns#getArrayType(java.lang.Class)
+		 */
+		@Override
+		public Class<?> getArrayType(Class<?> userType) {
+			throw new UnsupportedOperationException("Array types not supported");
+		}
+	}
+}

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

@@ -5,6 +5,7 @@ import java.util.Collections;
 import java.util.HashSet;
 
 import org.springframework.data.mapping.model.SimpleTypeHolder;
+import org.springframework.data.r2dbc.dialect.ArrayColumns.Unsupported;
 
 /**
  * Represents a dialect that is implemented by a particular database.
@@ -61,12 +62,11 @@ public interface Dialect {
 	LimitClause limit();
 
 	/**
-	 * Returns {@literal true} whether this dialect supports array-typed column. Collection-typed columns can map their
-	 * content to native array types.
+	 * Returns the array support object that describes how array-typed columns are supported by this dialect.
 	 *
-	 * @return {@literal true} whether this dialect supports array-typed columns.
+	 * @return the array support object that describes how array-typed columns are supported by this dialect.
 	 */
-	default boolean supportsArrayColumns() {
-		return false;
+	default ArrayColumns getArraySupport() {
+		return Unsupported.INSTANCE;
 	}
 }

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

@@ -1,15 +1,20 @@
 package org.springframework.data.r2dbc.dialect;
 
+import lombok.RequiredArgsConstructor;
+
 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;
 
+import org.springframework.data.mapping.model.SimpleTypeHolder;
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+
 /**
  * An SQL dialect for Postgres.
  *
@@ -18,7 +23,7 @@ import java.util.UUID;
 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));
+			Arrays.asList(UUID.class, URL.class, URI.class, InetAddress.class));
 
 	/**
 	 * Singleton instance.
@@ -57,6 +62,8 @@ public class PostgresDialect implements Dialect {
 		}
 	};
 
+	private final PostgresArrayColumns ARRAY_COLUMNS = new PostgresArrayColumns(getSimpleTypeHolder());
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.r2dbc.dialect.Dialect#getBindMarkersFactory()
@@ -95,10 +102,41 @@ public class PostgresDialect implements Dialect {
 
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.data.r2dbc.dialect.Dialect#supportsArrayColumns()
+	 * @see org.springframework.data.r2dbc.dialect.Dialect#getArraySupport()
 	 */
 	@Override
-	public boolean supportsArrayColumns() {
-		return true;
+	public ArrayColumns getArraySupport() {
+		return ARRAY_COLUMNS;
+	}
+
+	@RequiredArgsConstructor
+	static class PostgresArrayColumns implements ArrayColumns {
+
+		private final SimpleTypeHolder simpleTypes;
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.dialect.ArrayColumns#isSupported()
+		 */
+		@Override
+		public boolean isSupported() {
+			return true;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.dialect.ArrayColumns#getArrayType(java.lang.Class)
+		 */
+		@Override
+		public Class<?> getArrayType(Class<?> userType) {
+
+			Assert.notNull(userType, "Array component type must not be null");
+
+			if (!simpleTypes.isSimpleType(userType)) {
+				throw new IllegalArgumentException("Unsupported array type: " + ClassUtils.getQualifiedName(userType));
+			}
+
+			return ClassUtils.resolvePrimitiveIfNecessary(userType);
+		}
 	}
 }

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

@@ -38,6 +38,7 @@ 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.ArrayColumns;
 import org.springframework.data.r2dbc.dialect.BindMarker;
 import org.springframework.data.r2dbc.dialect.BindMarkers;
 import org.springframework.data.r2dbc.dialect.Dialect;
@@ -249,7 +250,7 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 	private Object getWriteValue(PersistentPropertyAccessor propertyAccessor, RelationalPersistentProperty property) {
 
 		TypeInformation<?> type = property.getTypeInformation();
-		Object value = relationalConverter.writeValue(propertyAccessor.getProperty(property), type);
+		Object value = propertyAccessor.getProperty(property);
 
 		if (type.isCollectionLike()) {
 
@@ -260,17 +261,28 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 				throw new InvalidDataAccessApiUsageException("Nested entities are not supported");
 			}
 
-			if (!dialect.supportsArrayColumns()) {
+			ArrayColumns arrayColumns = dialect.getArraySupport();
+
+			if (!arrayColumns.isSupported()) {
 
 				throw new InvalidDataAccessResourceUsageException(
 						"Dialect " + dialect.getClass().getName() + " does not support array columns");
 			}
 
-			if (!property.isArray()) {
+			return getArrayValue(arrayColumns, property, value);
+		}
 
-				Object zeroLengthArray = Array.newInstance(property.getActualType(), 0);
-				return relationalConverter.getConversionService().convert(value, zeroLengthArray.getClass());
-			}
+		return value;
+	}
+
+	private Object getArrayValue(ArrayColumns arrayColumns, RelationalPersistentProperty property, Object value) {
+
+		Class<?> targetType = arrayColumns.getArrayType(property.getActualType());
+
+		if (!property.isArray() || !property.getActualType().equals(targetType)) {
+
+			Object zeroLengthArray = Array.newInstance(targetType, 0);
+			return relationalConverter.getConversionService().convert(value, zeroLengthArray.getClass());
 		}
 
 		return value;

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

@@ -96,7 +96,8 @@ public class EntityRowMapper<T> implements BiFunction<Row, RowMetadata, T> {
 				return readEntityFrom(row, property);
 			}
 
-			return converter.readValue(row.get(prefix + property.getColumnName()), property.getTypeInformation());
+			Object value = row.get(prefix + property.getColumnName());
+			return converter.readValue(value, property.getTypeInformation());
 
 		} catch (Exception o_O) {
 			throw new MappingException(String.format("Could not read property %s from result set!", property), o_O);
@@ -156,9 +157,7 @@ public class EntityRowMapper<T> implements BiFunction<Row, RowMetadata, T> {
 			String column = prefix + property.getColumnName();
 
 			try {
-
-				Object value = converter.readValue(resultSet.get(column), property.getTypeInformation());
-				return converter.getConversionService().convert(value, parameter.getType().getType());
+				return converter.getConversionService().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);
 			}