Migrate R2DBC to read RowDocument.

Original pull request #1618
See #1554

spring-data-r2dbc/src/main/java/org/springframework/data/r2dbc/convert/MappingR2dbcConverter.java

@@ -18,6 +18,8 @@
 import io.r2dbc.spi.Blob;
 import io.r2dbc.spi.Clob;
 import io.r2dbc.spi.ColumnMetadata;
+import io.r2dbc.spi.Readable;
+import io.r2dbc.spi.ReadableMetadata;
 import io.r2dbc.spi.Row;
 import io.r2dbc.spi.RowMetadata;
 
@@ -53,6 +55,7 @@
 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.relational.domain.RowDocument;
 import org.springframework.data.util.TypeInformation;
 import org.springframework.lang.Nullable;
 import org.springframework.r2dbc.core.Parameter;
@@ -75,7 +78,8 @@ public class MappingR2dbcConverter extends MappingRelationalConverter implements
 	 */
 	public MappingR2dbcConverter(
 			MappingContext<? extends RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty> context) {
-		super((RelationalMappingContext) context, new R2dbcCustomConversions(R2dbcCustomConversions.STORE_CONVERSIONS, Collections.emptyList()));
+		super((RelationalMappingContext) context,
+				new R2dbcCustomConversions(R2dbcCustomConversions.STORE_CONVERSIONS, Collections.emptyList()));
 	}
 
 	/**
@@ -141,6 +145,54 @@ private <R> R read(RelationalPersistentEntity<R> entity, Row row, @Nullable RowM
 		return result;
 	}
 
+	@Override
+	public RowDocument toRowDocument(Class<?> type, Readable row, Iterable<? extends ReadableMetadata> metadata) {
+
+		RowDocument document = new RowDocument();
+		RelationalPersistentEntity<?> persistentEntity = getMappingContext().getPersistentEntity(type);
+
+		if (persistentEntity != null) {
+			captureRowValues(row, metadata, document, persistentEntity);
+		}
+
+		for (ReadableMetadata m : metadata) {
+
+			if (document.containsKey(m.getName())) {
+				continue;
+			}
+
+			document.put(m.getName(), row.get(m.getName()));
+		}
+
+		return document;
+	}
+
+	private static void captureRowValues(Readable row, Iterable<? extends ReadableMetadata> metadata,
+			RowDocument document, RelationalPersistentEntity<?> persistentEntity) {
+
+		for (RelationalPersistentProperty property : persistentEntity) {
+
+			String identifier = property.getColumnName().getReference();
+
+			if (property.isEntity() || !RowMetadataUtils.containsColumn(metadata, identifier)) {
+				continue;
+			}
+
+			Object value;
+			Class<?> propertyType = property.getType();
+
+			if (propertyType.equals(Clob.class)) {
+				value = row.get(identifier, Clob.class);
+			} else if (propertyType.equals(Blob.class)) {
+				value = row.get(identifier, Blob.class);
+			} else {
+				value = row.get(identifier);
+			}
+
+			document.put(identifier, value);
+		}
+	}
+
 	/**
 	 * Read a single value or a complete Entity from the {@link Row} passed as an argument.
 	 *

spring-data-r2dbc/src/main/java/org/springframework/data/r2dbc/convert/R2dbcConverter.java

@@ -15,6 +15,8 @@
  */
 package org.springframework.data.r2dbc.convert;
 
+import io.r2dbc.spi.Readable;
+import io.r2dbc.spi.ReadableMetadata;
 import io.r2dbc.spi.Row;
 import io.r2dbc.spi.RowMetadata;
 
@@ -29,6 +31,7 @@
 import org.springframework.data.relational.core.dialect.ArrayColumns;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
+import org.springframework.data.relational.domain.RowDocument;
 
 /**
  * Central R2DBC specific converter interface.
@@ -103,4 +106,15 @@ public interface R2dbcConverter
 	 */
 	<R> R read(Class<R> type, Row source, RowMetadata metadata);
 
+	/**
+	 * Create a flat {@link RowDocument} from a single {@link Readable Row or Stored Procedure output}.
+	 *
+	 * @param type the underlying entity type.
+	 * @param row the row or stored procedure output to retrieve data from.
+	 * @param metadata readable metadata.
+	 * @return the {@link RowDocument} containing the data.
+	 * @since 3.2
+	 */
+	RowDocument toRowDocument(Class<?> type, Readable row, Iterable<? extends ReadableMetadata> metadata);
+
 }

spring-data-r2dbc/src/main/java/org/springframework/data/r2dbc/convert/RowMetadataUtils.java

@@ -16,6 +16,7 @@
 package org.springframework.data.r2dbc.convert;
 
 import io.r2dbc.spi.ColumnMetadata;
+import io.r2dbc.spi.ReadableMetadata;
 import io.r2dbc.spi.RowMetadata;
 
 /**
@@ -34,10 +35,19 @@ class RowMetadataUtils {
 	 * @return {@code true} if the metadata contains the column {@code name}.
 	 */
 	public static boolean containsColumn(RowMetadata metadata, String name) {
+		return containsColumn(getColumnMetadata(metadata), name);
+	}
 
-		Iterable<? extends ColumnMetadata> columns = getColumnMetadata(metadata);
+	/**
+	 * Check whether the column {@code name} is contained in {@link RowMetadata}. The check happens case-insensitive.
+	 *
+	 * @param columns the metadata to inspect.
+	 * @param name column name.
+	 * @return {@code true} if the metadata contains the column {@code name}.
+	 */
+	public static boolean containsColumn(Iterable<? extends ReadableMetadata> columns, String name) {
 
-		for (ColumnMetadata columnMetadata : columns) {
+		for (ReadableMetadata columnMetadata : columns) {
 			if (name.equalsIgnoreCase(columnMetadata.getName())) {
 				return true;
 			}

spring-data-r2dbc/src/main/java/org/springframework/data/r2dbc/core/DefaultReactiveDataAccessStrategy.java

@@ -15,6 +15,8 @@
  */
 package org.springframework.data.r2dbc.core;
 
+import io.r2dbc.spi.Readable;
+import io.r2dbc.spi.ReadableMetadata;
 import io.r2dbc.spi.Row;
 import io.r2dbc.spi.RowMetadata;
 
@@ -43,6 +45,7 @@
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
 import org.springframework.data.relational.core.sql.SqlIdentifier;
+import org.springframework.data.relational.domain.RowDocument;
 import org.springframework.lang.Nullable;
 import org.springframework.r2dbc.core.Parameter;
 import org.springframework.r2dbc.core.PreparedOperation;
@@ -239,8 +242,7 @@ private Parameter getArrayValue(Parameter value, RelationalPersistentProperty pr
 			return Parameter.empty(targetArrayType);
 		}
 
-		return Parameter.fromOrEmpty(this.converter.getArrayValue(arrayColumns, property, value.getValue()),
-				actualType);
+		return Parameter.fromOrEmpty(this.converter.getArrayValue(arrayColumns, property, value.getValue()), actualType);
 	}
 
 	@Override
@@ -253,6 +255,11 @@ public <T> BiFunction<Row, RowMetadata, T> getRowMapper(Class<T> typeToRead) {
 		return new EntityRowMapper<>(typeToRead, this.converter);
 	}
 
+	@Override
+	public RowDocument toRowDocument(Class<?> type, Readable row, Iterable<? extends ReadableMetadata> metadata) {
+		return this.converter.toRowDocument(type, row, metadata);
+	}
+
 	@Override
 	public PreparedOperation<?> processNamedParameters(String query, NamedParameterProvider parameterProvider) {
 
@@ -289,6 +296,7 @@ public StatementMapper getStatementMapper() {
 		return this.statementMapper;
 	}
 
+	@Override
 	public R2dbcConverter getConverter() {
 		return this.converter;
 	}

spring-data-r2dbc/src/main/java/org/springframework/data/r2dbc/core/R2dbcEntityTemplate.java

@@ -45,6 +45,7 @@
 import org.springframework.data.mapping.PersistentPropertyAccessor;
 import org.springframework.data.mapping.callback.ReactiveEntityCallbacks;
 import org.springframework.data.mapping.context.MappingContext;
+import org.springframework.data.projection.EntityProjection;
 import org.springframework.data.projection.ProjectionInformation;
 import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
 import org.springframework.data.r2dbc.convert.R2dbcConverter;
@@ -66,6 +67,7 @@
 import org.springframework.data.relational.core.sql.Functions;
 import org.springframework.data.relational.core.sql.SqlIdentifier;
 import org.springframework.data.relational.core.sql.Table;
+import org.springframework.data.relational.domain.RowDocument;
 import org.springframework.data.util.ProxyUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.r2dbc.core.DatabaseClient;
@@ -95,6 +97,8 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
 
 	private final ReactiveDataAccessStrategy dataAccessStrategy;
 
+	private final R2dbcConverter converter;
+
 	private final MappingContext<? extends RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty> mappingContext;
 
 	private final SpelAwareProxyProjectionFactory projectionFactory;
@@ -116,7 +120,8 @@ public R2dbcEntityTemplate(ConnectionFactory connectionFactory) {
 		this.databaseClient = DatabaseClient.builder().connectionFactory(connectionFactory)
 				.bindMarkers(dialect.getBindMarkersFactory()).build();
 		this.dataAccessStrategy = new DefaultReactiveDataAccessStrategy(dialect);
-		this.mappingContext = dataAccessStrategy.getConverter().getMappingContext();
+		this.converter = dataAccessStrategy.getConverter();
+		this.mappingContext = converter.getMappingContext();
 		this.projectionFactory = new SpelAwareProxyProjectionFactory();
 	}
 
@@ -157,6 +162,7 @@ public R2dbcEntityTemplate(DatabaseClient databaseClient, ReactiveDataAccessStra
 
 		this.databaseClient = databaseClient;
 		this.dataAccessStrategy = strategy;
+		this.converter = dataAccessStrategy.getConverter();
 		this.mappingContext = strategy.getConverter().getMappingContext();
 		this.projectionFactory = new SpelAwareProxyProjectionFactory();
 	}
@@ -173,7 +179,7 @@ public ReactiveDataAccessStrategy getDataAccessStrategy() {
 
 	@Override
 	public R2dbcConverter getConverter() {
-		return this.dataAccessStrategy.getConverter();
+		return this.converter;
 	}
 
 	@Override
@@ -334,10 +340,10 @@ <T, P extends Publisher<T>> P doSelect(Query query, Class<?> entityClass, SqlIde
 		return (P) ((Flux<?>) result).concatMap(it -> maybeCallAfterConvert(it, tableName));
 	}
 
-	private <T> RowsFetchSpec<T> doSelect(Query query, Class<?> entityClass, SqlIdentifier tableName,
+	private <T> RowsFetchSpec<T> doSelect(Query query, Class<?> entityType, SqlIdentifier tableName,
 			Class<T> returnType) {
 
-		StatementMapper statementMapper = dataAccessStrategy.getStatementMapper().forType(entityClass);
+		StatementMapper statementMapper = dataAccessStrategy.getStatementMapper().forType(entityType);
 
 		StatementMapper.SelectSpec selectSpec = statementMapper //
 				.createSelect(tableName) //
@@ -362,7 +368,7 @@ private <T> RowsFetchSpec<T> doSelect(Query query, Class<?> entityClass, SqlIden
 
 		PreparedOperation<?> operation = statementMapper.getMappedObject(selectSpec);
 
-		return getRowsFetchSpec(databaseClient.sql(operation), entityClass, returnType);
+		return getRowsFetchSpec(databaseClient.sql(operation), entityType, returnType);
 	}
 
 	@Override
@@ -783,19 +789,26 @@ private <T> List<Expression> getSelectProjection(Table table, Query query, Class
 		return query.getColumns().stream().map(table::column).collect(Collectors.toList());
 	}
 
-	private <T> RowsFetchSpec<T> getRowsFetchSpec(DatabaseClient.GenericExecuteSpec executeSpec, Class<?> entityClass,
-			Class<T> returnType) {
+	private <T> RowsFetchSpec<T> getRowsFetchSpec(DatabaseClient.GenericExecuteSpec executeSpec, Class<?> entityType,
+			Class<T> resultType) {
 
-		boolean simpleType;
+		boolean simpleType = getConverter().isSimpleType(resultType);
 
 		BiFunction<Row, RowMetadata, T> rowMapper;
-		if (returnType.isInterface()) {
-			simpleType = getConverter().isSimpleType(entityClass);
-			rowMapper = dataAccessStrategy.getRowMapper(entityClass)
-					.andThen(o -> projectionFactory.createProjection(returnType, o));
+
+		if (simpleType) {
+			rowMapper = dataAccessStrategy.getRowMapper(resultType);
 		} else {
-			simpleType = getConverter().isSimpleType(returnType);
-			rowMapper = dataAccessStrategy.getRowMapper(returnType);
+
+			EntityProjection<T, ?> projection = converter.introspectProjection(resultType, entityType);
+
+			rowMapper = (row, rowMetadata) -> {
+
+				RowDocument document = dataAccessStrategy.toRowDocument(resultType, row, rowMetadata.getColumnMetadatas());
+
+				return projection.isProjection() ? converter.project(projection, document)
+						: converter.read(resultType, document);
+			};
 		}
 
 		// avoid top-level null values if the read type is a simple one (e.g. SELECT MAX(age) via Integer.class)

spring-data-r2dbc/src/main/java/org/springframework/data/r2dbc/core/ReactiveDataAccessStrategy.java

@@ -15,6 +15,8 @@
  */
 package org.springframework.data.r2dbc.core;
 
+import io.r2dbc.spi.Readable;
+import io.r2dbc.spi.ReadableMetadata;
 import io.r2dbc.spi.Row;
 import io.r2dbc.spi.RowMetadata;
 
@@ -25,6 +27,7 @@
 import org.springframework.data.r2dbc.mapping.OutboundRow;
 import org.springframework.data.relational.core.sql.IdentifierProcessing;
 import org.springframework.data.relational.core.sql.SqlIdentifier;
+import org.springframework.data.relational.domain.RowDocument;
 import org.springframework.lang.Nullable;
 import org.springframework.r2dbc.core.Parameter;
 import org.springframework.r2dbc.core.PreparedOperation;
@@ -82,6 +85,17 @@ public interface ReactiveDataAccessStrategy {
 	 */
 	<T> BiFunction<Row, RowMetadata, T> getRowMapper(Class<T> typeToRead);
 
+	/**
+	 * Create a flat {@link RowDocument} from a single {@link Readable Row or Stored Procedure output}.
+	 *
+	 * @param type the underlying entity type.
+	 * @param row the row or stored procedure output to retrieve data from.
+	 * @param metadata readable metadata.
+	 * @return the {@link RowDocument} containing the data.
+	 * @since 3.2
+	 */
+	RowDocument toRowDocument(Class<?> type, Readable row, Iterable<? extends ReadableMetadata> metadata);
+
 	/**
 	 * @param type
 	 * @return the table name for the {@link Class entity type}.

spring-data-r2dbc/src/test/java/org/springframework/data/r2dbc/core/ReactiveSelectOperationUnitTests.java

@@ -28,7 +28,6 @@
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-
 import org.springframework.data.annotation.Id;
 import org.springframework.data.r2dbc.dialect.PostgresDialect;
 import org.springframework.data.r2dbc.testing.StatementRecorder;
@@ -103,6 +102,30 @@ void shouldSelectAs() {
 		assertThat(statement.getSql()).isEqualTo("SELECT person.THE_NAME FROM person WHERE person.THE_NAME = $1");
 	}
 
+	@Test // gh-220
+	void shouldSelectAsWithColumnName() {
+
+		MockRowMetadata metadata = MockRowMetadata.builder()
+				.columnMetadata(MockColumnMetadata.builder().name("id").type(R2dbcType.INTEGER).build())
+				.columnMetadata(MockColumnMetadata.builder().name("a_different_name").type(R2dbcType.VARCHAR).build()).build();
+		MockResult result = MockResult.builder().row(MockRow.builder().identified("id", Object.class, "Walter")
+				.identified("a_different_name", Object.class, "Werner").metadata(metadata).build()).build();
+
+		recorder.addStubbing(s -> s.startsWith("SELECT"), result);
+
+		entityTemplate.select(Person.class) //
+				.as(PersonProjectionWithColumnName.class) //
+				.matching(query(where("name").is("Walter"))) //
+				.all() //
+				.as(StepVerifier::create) //
+				.assertNext(actual -> assertThat(actual.getName()).isEqualTo("Werner")) //
+				.verifyComplete();
+
+		StatementRecorder.RecordedStatement statement = recorder.getCreatedStatement(s -> s.startsWith("SELECT"));
+
+		assertThat(statement.getSql()).isEqualTo("SELECT person.* FROM person WHERE person.THE_NAME = $1");
+	}
+
 	@Test // gh-220
 	void shouldSelectFromTable() {
 
@@ -234,6 +257,21 @@ public void setName(String name) {
 		}
 	}
 
+	static class PersonProjectionWithColumnName {
+
+		@Id String id;
+
+		@Column("a_different_name") String name;
+
+		public String getName() {
+			return name;
+		}
+
+		public void setName(String name) {
+			this.name = name;
+		}
+	}
+
 	interface PersonProjection {
 
 		String getName();