Use Converter-based projection for R2DBC repository queries.

Previously, we instantiated the underlying entity. Now, we either read results directly into the result type or use a Map-backed projection.

Closes #1687

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

@@ -159,6 +159,22 @@ public interface R2dbcEntityOperations extends FluentR2dbcOperations {
 	 */
 	<T> RowsFetchSpec<T> query(PreparedOperation<?> operation, Class<T> entityClass) throws DataAccessException;
 
+	/**
+	 * Execute a query for a {@link RowsFetchSpec}, given {@link PreparedOperation}. Any provided bindings within
+	 * {@link PreparedOperation} are applied to the underlying {@link DatabaseClient}. The query is issued as-is without
+	 * additional pre-processing such as named parameter expansion. Results of the query are mapped onto
+	 * {@code entityClass}.
+	 *
+	 * @param operation   the prepared operation wrapping a SQL query and bind parameters.
+	 * @param entityClass the entity type must not be {@literal null}.
+	 * @param resultType  the returned entity, type must not be {@literal null}.
+	 * @return a {@link RowsFetchSpec} ready to materialize.
+	 * @throws DataAccessException if there is any problem issuing the execution.
+	 * @since 3.2.1
+	 */
+	<T> RowsFetchSpec<T> query(PreparedOperation<?> operation, Class<?> entityClass, Class<T> resultType)
+			throws DataAccessException;
+
 	/**
 	 * Execute a query for a {@link RowsFetchSpec}, given {@link PreparedOperation}. Any provided bindings within
 	 * {@link PreparedOperation} are applied to the underlying {@link DatabaseClient}. The query is issued as-is without
@@ -234,6 +250,9 @@ public interface R2dbcEntityOperations extends FluentR2dbcOperations {
 	<T> RowsFetchSpec<T> query(PreparedOperation<?> operation, Class<?> entityClass,
 			BiFunction<Row, RowMetadata, T> rowMapper) throws DataAccessException;
 
+	<T> RowsFetchSpec<T> getRowsFetchSpec(DatabaseClient.GenericExecuteSpec executeSpec, Class<?> entityType,
+			Class<T> resultType);
+
 	// -------------------------------------------------------------------------
 	// Methods dealing with entities
 	// -------------------------------------------------------------------------

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

@@ -419,11 +419,16 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
 
 	@Override
 	public <T> RowsFetchSpec<T> query(PreparedOperation<?> operation, Class<T> entityClass) {
+		return query(operation, entityClass, entityClass);
+	}
+
+	@Override
+	public <T> RowsFetchSpec<T> query(PreparedOperation<?> operation, Class<?> entityClass, Class<T> resultType) throws DataAccessException {
 
 		Assert.notNull(operation, "PreparedOperation must not be null");
 		Assert.notNull(entityClass, "Entity class must not be null");
 
-		return new EntityCallbackAdapter<>(getRowsFetchSpec(databaseClient.sql(operation), entityClass, entityClass),
+		return new EntityCallbackAdapter<>(getRowsFetchSpec(databaseClient.sql(operation), entityClass, resultType),
 				getTableNameOrEmpty(entityClass));
 	}
 
@@ -774,7 +779,7 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
 		return query.getColumns().stream().map(table::column).collect(Collectors.toList());
 	}
 
-	private <T> RowsFetchSpec<T> getRowsFetchSpec(DatabaseClient.GenericExecuteSpec executeSpec, Class<?> entityType,
+	public <T> RowsFetchSpec<T> getRowsFetchSpec(DatabaseClient.GenericExecuteSpec executeSpec, Class<?> entityType,
 			Class<T> resultType) {
 
 		boolean simpleType = getConverter().isSimpleType(resultType);

spring-data-r2dbc/src/main/java/org/springframework/data/r2dbc/repository/query/AbstractR2dbcQuery.java

@@ -90,13 +90,15 @@ public abstract class AbstractR2dbcQuery implements RepositoryQuery {
 		} else if (isExistsQuery()) {
 			fetchSpec = entityOperations.getDatabaseClient().sql(operation).map(row -> true);
 		} else {
-			fetchSpec = entityOperations.query(operation, resolveResultType(processor));
+			fetchSpec = entityOperations.query(operation, processor.getReturnedType()
+							.getDomainType(),
+					resolveResultType(processor));
 		}
 
 		R2dbcQueryExecution execution = new ResultProcessingExecution(getExecutionToWrap(processor.getReturnedType()),
 				new ResultProcessingConverter(processor, converter.getMappingContext(), instantiators));
 
-		return execution.execute(RowsFetchSpec.class.cast(fetchSpec));
+		return execution.execute((RowsFetchSpec) fetchSpec);
 	}
 
 	Class<?> resolveResultType(ResultProcessor resultProcessor) {
@@ -107,7 +109,7 @@ public abstract class AbstractR2dbcQuery implements RepositoryQuery {
 			return returnedType.getDomainType();
 		}
 
-		return returnedType.isProjecting() ? returnedType.getDomainType() : returnedType.getReturnedType();
+		return returnedType.getReturnedType();
 	}
 
 	private R2dbcQueryExecution getExecutionToWrap(ReturnedType returnedType) {

spring-data-r2dbc/src/test/java/org/springframework/data/r2dbc/repository/ConvertingR2dbcRepositoryIntegrationTests.java

@@ -105,7 +105,7 @@ public class ConvertingR2dbcRepositoryIntegrationTests {
 	}
 
 	@Test
-	public void shouldInsertAndReadItems() {
+	void shouldInsertAndReadItems() {
 
 		ConvertedEntity entity = new ConvertedEntity();
 		entity.name = "name";

spring-data-r2dbc/src/test/java/org/springframework/data/r2dbc/repository/ProjectingRepositoryIntegrationTests.java

@@ -0,0 +1,170 @@
+/*
+ * Copyright 2019-2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.r2dbc.repository;
+
+import static org.assertj.core.api.Assertions.*;
+
+import io.r2dbc.mssql.util.Assert;
+import io.r2dbc.spi.ConnectionFactory;
+import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
+
+import javax.sql.DataSource;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.ComponentScan;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.FilterType;
+import org.springframework.dao.DataAccessException;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.r2dbc.config.AbstractR2dbcConfiguration;
+import org.springframework.data.r2dbc.repository.config.EnableR2dbcRepositories;
+import org.springframework.data.r2dbc.testing.H2TestSupport;
+import org.springframework.data.relational.core.mapping.Table;
+import org.springframework.data.repository.reactive.ReactiveCrudRepository;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.lang.Nullable;
+import org.springframework.test.context.junit.jupiter.SpringExtension;
+
+/**
+ * Integration tests projections.
+ *
+ * @author Mark Paluch
+ */
+@ExtendWith(SpringExtension.class)
+public class ProjectingRepositoryIntegrationTests {
+
+	@Autowired
+	private ImmutableObjectRepository repository;
+	private JdbcTemplate jdbc;
+
+	@Configuration
+	@EnableR2dbcRepositories(
+			includeFilters = @ComponentScan.Filter(value = ImmutableObjectRepository.class, type = FilterType.ASSIGNABLE_TYPE),
+			considerNestedRepositories = true)
+	static class TestConfiguration extends AbstractR2dbcConfiguration {
+		@Override
+		public ConnectionFactory connectionFactory() {
+			return H2TestSupport.createConnectionFactory();
+		}
+
+	}
+
+	@BeforeEach
+	void before() {
+
+		this.jdbc = new JdbcTemplate(createDataSource());
+
+		try {
+			this.jdbc.execute("DROP TABLE immutable_non_null");
+		}
+		catch (DataAccessException e) {
+		}
+
+		this.jdbc.execute("CREATE TABLE immutable_non_null (id serial PRIMARY KEY, name varchar(255), email varchar(255))");
+		this.jdbc.execute("INSERT INTO immutable_non_null VALUES (42, 'Walter', 'heisenberg@the-white-family.com')");
+	}
+
+	/**
+	 * Creates a {@link DataSource} to be used in this test.
+	 *
+	 * @return the {@link DataSource} to be used in this test.
+	 */
+	protected DataSource createDataSource() {
+		return H2TestSupport.createDataSource();
+	}
+
+	/**
+	 * Creates a {@link ConnectionFactory} to be used in this test.
+	 *
+	 * @return the {@link ConnectionFactory} to be used in this test.
+	 */
+	protected ConnectionFactory createConnectionFactory() {
+		return H2TestSupport.createConnectionFactory();
+	}
+
+	@Test
+		// GH-1687
+	void shouldApplyProjectionDirectly() {
+
+		repository.findProjectionByEmail("heisenberg@the-white-family.com") //
+				.as(StepVerifier::create) //
+				.consumeNextWith(actual -> {
+					assertThat(actual.getName()).isEqualTo("Walter");
+				}).verifyComplete();
+	}
+
+	@Test
+		// GH-1687
+	void shouldApplyEntityQueryProjectionDirectly() {
+
+		repository.findAllByEmail("heisenberg@the-white-family.com") //
+				.as(StepVerifier::create) //
+				.consumeNextWith(actual -> {
+					assertThat(actual.getName()).isEqualTo("Walter");
+					assertThat(actual).isInstanceOf(ImmutableNonNullEntity.class);
+				}).verifyComplete();
+	}
+
+	interface ImmutableObjectRepository extends ReactiveCrudRepository<ImmutableNonNullEntity, Integer> {
+
+		Flux<ProjectionOnNonNull> findProjectionByEmail(String email);
+
+		Flux<Person> findAllByEmail(String email);
+
+	}
+
+	@Table("immutable_non_null")
+	static class ImmutableNonNullEntity implements Person {
+
+		final @Nullable
+		@Id Integer id;
+		final String name;
+		final String email;
+
+		ImmutableNonNullEntity(@Nullable Integer id, String name, String email) {
+
+			Assert.notNull(name, "Name must not be null");
+			Assert.notNull(email, "Email must not be null");
+
+			this.id = id;
+			this.name = name;
+			this.email = email;
+		}
+
+		@Override
+		public String getName() {
+			return name;
+		}
+	}
+
+	interface Person {
+
+		String getName();
+
+	}
+
+	interface ProjectionOnNonNull {
+
+		String getName();
+
+	}
+
+}

spring-data-r2dbc/src/test/java/org/springframework/data/r2dbc/repository/query/StringBasedR2dbcQueryUnitTests.java

@@ -266,12 +266,18 @@ public class StringBasedR2dbcQueryUnitTests {
 		verifyNoMoreInteractions(bindTarget);
 	}
 
-	@Test // gh-475
-	void usesDomainTypeForInterfaceProjectionResultMapping() {
+	@Test
+		// gh-475, GH-1687
+	void usesProjectionTypeForInterfaceProjectionResultMapping() {
 
 		StringBasedR2dbcQuery query = getQueryMethod("findAsInterfaceProjection");
 
-		assertThat(query.resolveResultType(query.getQueryMethod().getResultProcessor())).isEqualTo(Person.class);
+		assertThat(query.getQueryMethod().getResultProcessor().getReturnedType()
+				.getReturnedType()).isEqualTo(PersonProjection.class);
+		assertThat(query.getQueryMethod().getResultProcessor().getReturnedType()
+				.getDomainType()).isEqualTo(Person.class);
+		assertThat(query.resolveResultType(query.getQueryMethod()
+				.getResultProcessor())).isEqualTo(PersonProjection.class);
 	}
 
 	@Test // gh-475