Fix `R2dbcEntityTemplate.getRowsFetchSpec(…)` to use the correct `Converter` for result type conversion.

Closes #1723

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

@@ -93,6 +93,7 @@ import org.springframework.util.Assert;
  * @author Jens Schauder
  * @author Jose Luis Leon
  * @author Robert Heim
+ * @author Sebastian Wieland
  * @since 1.1
  */
 public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAware, ApplicationContextAware {
@@ -821,10 +822,10 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
 
 		// Bridge-code: Consider Converter<Row, T> until we have fully migrated to RowDocument
 		if (converter instanceof AbstractRelationalConverter relationalConverter
-				&& relationalConverter.getConversions().hasCustomReadTarget(Row.class, entityType)) {
+				&& relationalConverter.getConversions().hasCustomReadTarget(Row.class, resultType)) {
 
 			ConversionService conversionService = relationalConverter.getConversionService();
-			rowMapper = (row, rowMetadata) -> (T) conversionService.convert(row, entityType);
+			rowMapper = (row, rowMetadata) -> (T) conversionService.convert(row, resultType);
 		} else if (simpleType) {
 			rowMapper = dataAccessStrategy.getRowMapper(resultType);
 		} else {

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

@@ -19,6 +19,7 @@ import static org.assertj.core.api.Assertions.*;
 
 import io.r2dbc.spi.ConnectionFactory;
 import io.r2dbc.spi.Row;
+import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
 import java.util.Arrays;
@@ -51,6 +52,7 @@ import org.springframework.test.context.junit.jupiter.SpringExtension;
  * Integration tests for {@link ConvertedRepository} that uses {@link Converter}s on entity-level.
  *
  * @author Mark Paluch
+ * @author Sebastian Wieland
  */
 @ExtendWith(SpringExtension.class)
 public class ConvertingR2dbcRepositoryIntegrationTests {
@@ -122,8 +124,26 @@ public class ConvertingR2dbcRepositoryIntegrationTests {
 				}).verifyComplete();
 	}
 
-	interface ConvertedRepository extends ReactiveCrudRepository<ConvertedEntity, Integer> {
+	@Test
+	void shouldNotUseConverterForCountQueries() {
+		ConvertedEntity entity = new ConvertedEntity();
+		entity.name = "name";
 
+		repository.save(entity) //
+				.as(StepVerifier::create) //
+				.expectNextCount(1) //
+				.verifyComplete();
+
+		repository.countWithCustomQuery() //
+				.as(StepVerifier::create) //
+				.consumeNextWith(actual -> {
+					assertThat(actual).isEqualTo(1L);
+				}).verifyComplete();
+	}
+
+	interface ConvertedRepository extends ReactiveCrudRepository<ConvertedEntity, Integer> {
+		@Query("SELECT COUNT(*) FROM CONVERTED_ENTITY")
+		Mono<Long> countWithCustomQuery();
 	}
 
 	static class ConvertedEntity {