Performance improvements.

Introduce caching for configured RowMapper/ResultSetExtractor.

We now create RowMapper/ResultSetExtractor instances only once if they are considered static configuration.
A configured RowMapper ref/class is always static.
A configured ResultSetExtractor ref/class is static when the extractor does not accept a RowMapper or if the RowMapper is configured.

Default mappers or projection-specific ones require ResultSetExtractor recreation of the ResultSetExtractor is configured as Class.

Reuse TypeInformation as much as possible to avoid Class -> TypeInformation conversion.
Introduce LRU cache for DefaultAggregatePath to avoid PersistentPropertyPath lookups.
Introduce best-effort quoted cache for SqlIdentifier to avoid excessive string object creation.

Closes #1721
Original pull request #1722

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/EntityRowMapper.java

@@ -79,8 +79,8 @@ public class EntityRowMapper<T> implements RowMapper<T> {
 		RowDocument document = RowDocumentResultSetExtractor.toRowDocument(resultSet);
 
 		return identifier == null //
-				? converter.readAndResolve(entity.getType(), document) //
-				: converter.readAndResolve(entity.getType(), document, identifier);
+				? converter.readAndResolve(entity.getTypeInformation(), document, Identifier.empty()) //
+				: converter.readAndResolve(entity.getTypeInformation(), document, identifier);
 	}
 
 }

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/JdbcConverter.java

@@ -26,6 +26,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;
 
 /**
@@ -47,7 +48,21 @@ public interface JdbcConverter extends RelationalConverter {
 	 * @return The converted value wrapped in a {@link JdbcValue}. Guaranteed to be not {@literal null}.
 	 * @since 2.4
 	 */
-	JdbcValue writeJdbcValue(@Nullable Object value, Class<?> type, SQLType sqlType);
+	default JdbcValue writeJdbcValue(@Nullable Object value, Class<?> type, SQLType sqlType) {
+		return writeJdbcValue(value, TypeInformation.of(type), sqlType);
+	}
+
+	/**
+	 * Convert a property value into a {@link JdbcValue} that contains the converted value and information how to bind it
+	 * to JDBC parameters.
+	 *
+	 * @param value a value as it is used in the object model. May be {@code null}.
+	 * @param type {@link TypeInformation} into which the value is to be converted. Must not be {@code null}.
+	 * @param sqlType the {@link SQLType} to be used if non is specified by a converter.
+	 * @return The converted value wrapped in a {@link JdbcValue}. Guaranteed to be not {@literal null}.
+	 * @since 3.2.6
+	 */
+	JdbcValue writeJdbcValue(@Nullable Object value, TypeInformation<?> type, SQLType sqlType);
 
 	/**
 	 * Read the current row from {@link ResultSet} to an {@link RelationalPersistentEntity#getType() entity}.
@@ -84,7 +99,7 @@ public interface JdbcConverter extends RelationalConverter {
 	default <T> T mapRow(PersistentPropertyPathExtension path, ResultSet resultSet, Identifier identifier, Object key) {
 
 		try {
-			return (T) readAndResolve(path.getRequiredLeafEntity().getType(),
+			return (T) readAndResolve(path.getRequiredLeafEntity().getTypeInformation(),
 					RowDocumentResultSetExtractor.toRowDocument(resultSet), identifier);
 		} catch (SQLException e) {
 			throw new RuntimeException(e);
@@ -118,7 +133,23 @@ public interface JdbcConverter extends RelationalConverter {
 	 * @since 3.2
 	 * @see #read(Class, RowDocument)
 	 */
-	<R> R readAndResolve(Class<R> type, RowDocument source, Identifier identifier);
+	default <R> R readAndResolve(Class<R> type, RowDocument source, Identifier identifier) {
+		return readAndResolve(TypeInformation.of(type), source, identifier);
+	}
+
+	/**
+	 * Read a {@link RowDocument} into the requested {@link TypeInformation aggregate type} and resolve references by
+	 * looking these up from {@link RelationResolver}.
+	 *
+	 * @param type target aggregate type.
+	 * @param source source {@link RowDocument}.
+	 * @param identifier identifier chain.
+	 * @return the converted object.
+	 * @param <R> aggregate type.
+	 * @since 3.2.6
+	 * @see #read(Class, RowDocument)
+	 */
+	<R> R readAndResolve(TypeInformation<R> type, RowDocument source, Identifier identifier);
 
 	/**
 	 * The type to be used to store this property in the database. Multidimensional arrays are unwrapped to reflect a
@@ -126,7 +157,7 @@ public interface JdbcConverter extends RelationalConverter {
 	 *
 	 * @return a {@link Class} that is suitable for usage with JDBC drivers.
 	 * @see org.springframework.data.jdbc.support.JdbcUtil#targetSqlTypeFor(Class)
-	 * @since 2.0
+	 * @since 2.0 TODO: Introduce variant returning TypeInformation.
 	 */
 	Class<?> getColumnType(RelationalPersistentProperty property);
 

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/MapEntityRowMapper.java

@@ -64,7 +64,7 @@ class MapEntityRowMapper<T> implements RowMapper<Map.Entry<Object, T>> {
 	@SuppressWarnings("unchecked")
 	private T mapEntity(RowDocument document, Object key) {
 
-		return (T) converter.readAndResolve(path.getRequiredLeafEntity().getType(), document,
+		return (T) converter.readAndResolve(path.getRequiredLeafEntity().getTypeInformation(), document,
 				identifier.withPart(keyColumn, key, Object.class));
 	}
 }

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/MappingJdbcConverter.java

@@ -231,14 +231,14 @@ public class MappingJdbcConverter extends MappingRelationalConverter implements
 	}
 
 	@Override
-	public JdbcValue writeJdbcValue(@Nullable Object value, Class<?> columnType, SQLType sqlType) {
+	public JdbcValue writeJdbcValue(@Nullable Object value, TypeInformation<?> columnType, SQLType sqlType) {
 
 		JdbcValue jdbcValue = tryToConvertToJdbcValue(value);
 		if (jdbcValue != null) {
 			return jdbcValue;
 		}
 
-		Object convertedValue = writeValue(value, TypeInformation.of(columnType));
+		Object convertedValue = writeValue(value, columnType);
 
 		if (convertedValue == null || !convertedValue.getClass().isArray()) {
 
@@ -275,7 +275,7 @@ public class MappingJdbcConverter extends MappingRelationalConverter implements
 
 	@SuppressWarnings("unchecked")
 	@Override
-	public <R> R readAndResolve(Class<R> type, RowDocument source, Identifier identifier) {
+	public <R> R readAndResolve(TypeInformation<R> type, RowDocument source, Identifier identifier) {
 
 		RelationalPersistentEntity<R> entity = (RelationalPersistentEntity<R>) getMappingContext()
 				.getRequiredPersistentEntity(type);

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/repository/query/AbstractJdbcQuery.java

@@ -18,6 +18,7 @@ package org.springframework.data.jdbc.repository.query;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.List;
+import java.util.function.Supplier;
 import java.util.stream.Stream;
 
 import org.springframework.core.convert.converter.Converter;
@@ -83,9 +84,9 @@ public abstract class AbstractJdbcQuery implements RepositoryQuery {
 	 */
 	@Deprecated(since = "3.1", forRemoval = true)
 	// a better name would be createReadingQueryExecution
-	protected JdbcQueryExecution<?> getQueryExecution(JdbcQueryMethod queryMethod,
+	JdbcQueryExecution<?> getQueryExecution(JdbcQueryMethod queryMethod,
 			@Nullable ResultSetExtractor<?> extractor, RowMapper<?> rowMapper) {
-		return createReadingQueryExecution(extractor, rowMapper);
+		return createReadingQueryExecution(extractor, () -> rowMapper);
 	}
 
 	/**
@@ -96,21 +97,21 @@ public abstract class AbstractJdbcQuery implements RepositoryQuery {
 	 * @param rowMapper must not be {@literal null}.
 	 * @return a JdbcQueryExecution appropriate for {@literal queryMethod}. Guaranteed to be not {@literal null}.
 	 */
-	protected JdbcQueryExecution<?> createReadingQueryExecution(@Nullable ResultSetExtractor<?> extractor,
-			RowMapper<?> rowMapper) {
+	JdbcQueryExecution<?> createReadingQueryExecution(@Nullable ResultSetExtractor<?> extractor,
+			Supplier<RowMapper<?>> rowMapper) {
 
 		if (getQueryMethod().isCollectionQuery()) {
-			return extractor != null ? createSingleReadingQueryExecution(extractor) : collectionQuery(rowMapper);
+			return extractor != null ? createSingleReadingQueryExecution(extractor) : collectionQuery(rowMapper.get());
 		}
 
 		if (getQueryMethod().isStreamQuery()) {
-			return extractor != null ? createSingleReadingQueryExecution(extractor) : streamQuery(rowMapper);
+			return extractor != null ? createSingleReadingQueryExecution(extractor) : streamQuery(rowMapper.get());
 		}
 
-		return extractor != null ? createSingleReadingQueryExecution(extractor) : singleObjectQuery(rowMapper);
+		return extractor != null ? createSingleReadingQueryExecution(extractor) : singleObjectQuery(rowMapper.get());
 	}
 
-	protected JdbcQueryExecution<Object> createModifyingQueryExecutor() {
+	JdbcQueryExecution<Object> createModifyingQueryExecutor() {
 
 		return (query, parameters) -> {
 

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/repository/query/JdbcParameters.java

@@ -0,0 +1,99 @@
+/*
+ * Copyright 2018-2024 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.jdbc.repository.query;
+
+import java.sql.SQLType;
+import java.util.List;
+
+import org.springframework.core.MethodParameter;
+import org.springframework.data.jdbc.core.convert.JdbcColumnTypes;
+import org.springframework.data.jdbc.support.JdbcUtil;
+import org.springframework.data.relational.repository.query.RelationalParameters;
+import org.springframework.data.repository.query.Parameter;
+import org.springframework.data.repository.query.ParametersSource;
+import org.springframework.data.util.Lazy;
+import org.springframework.data.util.TypeInformation;
+
+/**
+ * Custom extension of {@link RelationalParameters}.
+ *
+ * @author Mark Paluch
+ * @since 3.2.6
+ */
+public class JdbcParameters extends RelationalParameters {
+
+	/**
+	 * Creates a new {@link JdbcParameters} instance from the given {@link ParametersSource}.
+	 *
+	 * @param parametersSource must not be {@literal null}.
+	 */
+	public JdbcParameters(ParametersSource parametersSource) {
+		super(parametersSource,
+				methodParameter -> new JdbcParameter(methodParameter, parametersSource.getDomainTypeInformation()));
+	}
+
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	private JdbcParameters(List<JdbcParameter> parameters) {
+		super((List) parameters);
+	}
+
+	@Override
+	public JdbcParameter getParameter(int index) {
+		return (JdbcParameter) super.getParameter(index);
+	}
+
+	@Override
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	protected JdbcParameters createFrom(List<RelationalParameter> parameters) {
+		return new JdbcParameters((List) parameters);
+	}
+
+	/**
+	 * Custom {@link Parameter} implementation.
+	 *
+	 * @author Mark Paluch
+	 * @author Chirag Tailor
+	 */
+	public static class JdbcParameter extends RelationalParameter {
+
+		private final SQLType sqlType;
+		private final Lazy<SQLType> actualSqlType;
+
+		/**
+		 * Creates a new {@link RelationalParameter}.
+		 *
+		 * @param parameter must not be {@literal null}.
+		 */
+		JdbcParameter(MethodParameter parameter, TypeInformation<?> domainType) {
+			super(parameter, domainType);
+
+			TypeInformation<?> typeInformation = getTypeInformation();
+
+			sqlType = JdbcUtil.targetSqlTypeFor(JdbcColumnTypes.INSTANCE.resolvePrimitiveType(typeInformation.getType()));
+
+			actualSqlType = Lazy.of(() -> JdbcUtil
+					.targetSqlTypeFor(JdbcColumnTypes.INSTANCE.resolvePrimitiveType(typeInformation.getActualType().getType())));
+		}
+
+		public SQLType getSqlType() {
+			return sqlType;
+		}
+
+		public SQLType getActualSqlType() {
+			return actualSqlType.get();
+		}
+	}
+}

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/repository/query/JdbcQueryMethod.java

@@ -40,6 +40,7 @@ import org.springframework.jdbc.core.RowMapper;
 import org.springframework.lang.Nullable;
 import org.springframework.util.ClassUtils;
 import org.springframework.util.ConcurrentReferenceHashMap;
+import org.springframework.util.ObjectUtils;
 import org.springframework.util.StringUtils;
 
 /**
@@ -59,6 +60,7 @@ public class JdbcQueryMethod extends QueryMethod {
 	private final Map<Class<? extends Annotation>, Optional<Annotation>> annotationCache;
 	private final NamedQueries namedQueries;
 	private @Nullable RelationalEntityMetadata<?> metadata;
+	private final boolean modifyingQuery;
 
 	// TODO: Remove NamedQueries and put it into JdbcQueryLookupStrategy
 	public JdbcQueryMethod(Method method, RepositoryMetadata metadata, ProjectionFactory factory,
@@ -70,11 +72,12 @@ public class JdbcQueryMethod extends QueryMethod {
 		this.method = method;
 		this.mappingContext = mappingContext;
 		this.annotationCache = new ConcurrentReferenceHashMap<>();
+		this.modifyingQuery = AnnotationUtils.findAnnotation(method, Modifying.class) != null;
 	}
 
 	@Override
 	protected Parameters<?, ?> createParameters(ParametersSource parametersSource) {
-		return new RelationalParameters(parametersSource);
+		return new JdbcParameters(parametersSource);
 	}
 
 	@Override
@@ -108,8 +111,8 @@ public class JdbcQueryMethod extends QueryMethod {
 	}
 
 	@Override
-	public RelationalParameters getParameters() {
-		return (RelationalParameters) super.getParameters();
+	public JdbcParameters getParameters() {
+		return (JdbcParameters) super.getParameters();
 	}
 
 	/**
@@ -124,6 +127,17 @@ public class JdbcQueryMethod extends QueryMethod {
 		return StringUtils.hasText(annotatedValue) ? annotatedValue : getNamedQuery();
 	}
 
+	String getRequiredQuery() {
+
+		String query = getDeclaredQuery();
+
+		if (ObjectUtils.isEmpty(query)) {
+			throw new IllegalStateException(String.format("No query specified on %s", getName()));
+		}
+
+		return query;
+	}
+
 	/**
 	 * Returns the annotated query if it exists.
 	 *
@@ -210,7 +224,7 @@ public class JdbcQueryMethod extends QueryMethod {
 	 */
 	@Override
 	public boolean isModifyingQuery() {
-		return AnnotationUtils.findAnnotation(method, Modifying.class) != null;
+		return modifyingQuery;
 	}
 
 	@SuppressWarnings("unchecked")

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/repository/query/PartTreeJdbcQuery.java

@@ -21,7 +21,9 @@ import java.sql.ResultSet;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
+import java.util.function.Function;
 import java.util.function.LongSupplier;
+import java.util.function.Supplier;
 
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.data.domain.Pageable;
@@ -29,6 +31,7 @@ import org.springframework.data.domain.Slice;
 import org.springframework.data.domain.SliceImpl;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.jdbc.core.convert.JdbcConverter;
+import org.springframework.data.relational.core.conversion.RelationalConverter;
 import org.springframework.data.relational.core.dialect.Dialect;
 import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 import org.springframework.data.relational.repository.query.RelationalEntityMetadata;
@@ -39,6 +42,7 @@ import org.springframework.data.repository.query.ResultProcessor;
 import org.springframework.data.repository.query.ReturnedType;
 import org.springframework.data.repository.query.parser.PartTree;
 import org.springframework.data.support.PageableExecutionUtils;
+import org.springframework.data.util.Lazy;
 import org.springframework.jdbc.core.ResultSetExtractor;
 import org.springframework.jdbc.core.RowMapper;
 import org.springframework.jdbc.core.namedparam.NamedParameterJdbcOperations;
@@ -61,7 +65,7 @@ public class PartTreeJdbcQuery extends AbstractJdbcQuery {
 	private final Parameters<?, ?> parameters;
 	private final Dialect dialect;
 	private final JdbcConverter converter;
-	private final RowMapperFactory rowMapperFactory;
+	private final CachedRowMapperFactory cachedRowMapperFactory;
 	private final PartTree tree;
 
 	/**
@@ -105,12 +109,12 @@ public class PartTreeJdbcQuery extends AbstractJdbcQuery {
 		this.parameters = queryMethod.getParameters();
 		this.dialect = dialect;
 		this.converter = converter;
-		this.rowMapperFactory = rowMapperFactory;
 
-		this.tree = new PartTree(queryMethod.getName(), queryMethod.getResultProcessor()
-				.getReturnedType().getDomainType());
+		this.tree = new PartTree(queryMethod.getName(), queryMethod.getResultProcessor().getReturnedType().getDomainType());
 		JdbcQueryCreator.validate(this.tree, this.parameters, this.converter.getMappingContext());
 
+		this.cachedRowMapperFactory = new CachedRowMapperFactory(tree, rowMapperFactory, converter,
+				queryMethod.getResultProcessor());
 	}
 
 	private Sort getDynamicSort(RelationalParameterAccessor accessor) {
@@ -134,18 +138,9 @@ public class PartTreeJdbcQuery extends AbstractJdbcQuery {
 			RelationalParametersParameterAccessor accessor) {
 
 		ResultSetExtractor<Boolean> extractor = tree.isExistsProjection() ? (ResultSet::next) : null;
-
-		RowMapper<Object> rowMapper;
-
-		if (tree.isCountProjection() || tree.isExistsProjection()) {
-			rowMapper = rowMapperFactory.create(resolveTypeToRead(processor));
-		} else {
-
-			Converter<Object, Object> resultProcessingConverter = new ResultProcessingConverter(processor,
-					this.converter.getMappingContext(), this.converter.getEntityInstantiators());
-			rowMapper = new ConvertingRowMapper<>(rowMapperFactory.create(processor.getReturnedType().getDomainType()),
-					resultProcessingConverter);
-		}
+		Supplier<RowMapper<?>> rowMapper = parameters.hasDynamicProjection()
+				? () -> cachedRowMapperFactory.getRowMapper(processor)
+				: cachedRowMapperFactory;
 
 		JdbcQueryExecution<?> queryExecution = getJdbcQueryExecution(extractor, rowMapper);
 
@@ -174,7 +169,7 @@ public class PartTreeJdbcQuery extends AbstractJdbcQuery {
 		return queryExecution;
 	}
 
-	protected ParametrizedQuery createQuery(RelationalParametersParameterAccessor accessor, ReturnedType returnedType) {
+	ParametrizedQuery createQuery(RelationalParametersParameterAccessor accessor, ReturnedType returnedType) {
 
 		RelationalEntityMetadata<?> entityMetadata = getQueryMethod().getEntityInformation();
 
@@ -183,10 +178,11 @@ public class PartTreeJdbcQuery extends AbstractJdbcQuery {
 		return queryCreator.createQuery(getDynamicSort(accessor));
 	}
 
-	private JdbcQueryExecution<?> getJdbcQueryExecution(@Nullable ResultSetExtractor<Boolean> extractor, RowMapper<Object> rowMapper) {
+	private JdbcQueryExecution<?> getJdbcQueryExecution(@Nullable ResultSetExtractor<Boolean> extractor,
+			Supplier<RowMapper<?>> rowMapper) {
 
 		if (getQueryMethod().isPageQuery() || getQueryMethod().isSliceQuery()) {
-			return collectionQuery(rowMapper);
+			return collectionQuery(rowMapper.get());
 		} else {
 
 			if (getQueryMethod().isModifyingQuery()) {
@@ -259,4 +255,45 @@ public class PartTreeJdbcQuery extends AbstractJdbcQuery {
 		}
 
 	}
+
+	/**
+	 * Cached implementation of {@link RowMapper} suppler providing either a cached variant of the RowMapper or creating a
+	 * new one when using dynamic projections.
+	 */
+	class CachedRowMapperFactory implements Supplier<RowMapper<?>> {
+
+		private final Lazy<RowMapper<?>> rowMapper;
+		private final Function<ResultProcessor, RowMapper<?>> rowMapperFunction;
+
+		public CachedRowMapperFactory(PartTree tree, RowMapperFactory rowMapperFactory, RelationalConverter converter,
+				ResultProcessor defaultResultProcessor) {
+
+			this.rowMapperFunction = processor -> {
+
+				if (tree.isCountProjection() || tree.isExistsProjection()) {
+					return rowMapperFactory.create(resolveTypeToRead(processor));
+				}
+				Converter<Object, Object> resultProcessingConverter = new ResultProcessingConverter(processor,
+						converter.getMappingContext(), converter.getEntityInstantiators());
+				return new ConvertingRowMapper<>(rowMapperFactory.create(processor.getReturnedType().getDomainType()),
+						resultProcessingConverter);
+			};
+
+			this.rowMapper = Lazy.of(() -> this.rowMapperFunction.apply(defaultResultProcessor));
+		}
+
+		@Override
+		public RowMapper<?> get() {
+			return getRowMapper();
+		}
+
+		public RowMapper<?> getRowMapper() {
+			return rowMapper.get();
+		}
+
+		public RowMapper<?> getRowMapper(ResultProcessor resultProcessor) {
+			return rowMapperFunction.apply(resultProcessor);
+		}
+
+	}
 }

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/repository/query/StringBasedJdbcQuery.java

@@ -22,10 +22,13 @@ import java.sql.SQLType;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.function.Supplier;
 
+import org.springframework.beans.BeanInstantiationException;
 import org.springframework.beans.BeanUtils;
 import org.springframework.beans.factory.BeanFactory;
-import org.springframework.core.convert.converter.Converter;
 import org.springframework.data.jdbc.core.convert.JdbcColumnTypes;
 import org.springframework.data.jdbc.core.convert.JdbcConverter;
 import org.springframework.data.jdbc.core.mapping.JdbcValue;
@@ -40,6 +43,7 @@ import org.springframework.data.repository.query.QueryMethodEvaluationContextPro
 import org.springframework.data.repository.query.ResultProcessor;
 import org.springframework.data.repository.query.SpelEvaluator;
 import org.springframework.data.repository.query.SpelQueryContext;
+import org.springframework.data.util.Lazy;
 import org.springframework.data.util.TypeInformation;
 import org.springframework.jdbc.core.ResultSetExtractor;
 import org.springframework.jdbc.core.RowMapper;
@@ -48,6 +52,7 @@ import org.springframework.jdbc.core.namedparam.NamedParameterJdbcOperations;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
+import org.springframework.util.ConcurrentReferenceHashMap;
 import org.springframework.util.ObjectUtils;
 
 /**
@@ -70,8 +75,13 @@ public class StringBasedJdbcQuery extends AbstractJdbcQuery {
 	private static final String PARAMETER_NEEDS_TO_BE_NAMED = "For queries with named parameters you need to provide names for method parameters; Use @Param for query method parameters, or when on Java 8+ use the javac flag -parameters";
 	private final JdbcConverter converter;
 	private final RowMapperFactory rowMapperFactory;
+	private final SpelEvaluator spelEvaluator;
+	private final boolean containsSpelExpressions;
+	private final String query;
 	private BeanFactory beanFactory;
-	private final QueryMethodEvaluationContextProvider evaluationContextProvider;
+
+	private final CachedRowMapperFactory cachedRowMapperFactory;
+	private final CachedResultSetExtractorFactory cachedResultSetExtractorFactory;
 
 	/**
 	 * Creates a new {@link StringBasedJdbcQuery} for the given {@link JdbcQueryMethod}, {@link RelationalMappingContext}
@@ -106,7 +116,6 @@ public class StringBasedJdbcQuery extends AbstractJdbcQuery {
 
 		this.converter = converter;
 		this.rowMapperFactory = rowMapperFactory;
-		this.evaluationContextProvider = evaluationContextProvider;
 
 		if (queryMethod.isSliceQuery()) {
 			throw new UnsupportedOperationException(
@@ -122,6 +131,19 @@ public class StringBasedJdbcQuery extends AbstractJdbcQuery {
 			throw new UnsupportedOperationException(
 					"Queries with Limit are not supported using string-based queries; Offending method: " + queryMethod);
 		}
+
+		this.cachedRowMapperFactory = new CachedRowMapperFactory(
+				() -> rowMapperFactory.create(queryMethod.getResultProcessor().getReturnedType().getReturnedType()));
+		this.cachedResultSetExtractorFactory = new CachedResultSetExtractorFactory(
+				this.cachedRowMapperFactory::getRowMapper);
+
+		SpelQueryContext.EvaluatingSpelQueryContext queryContext = SpelQueryContext
+				.of((counter, expression) -> String.format("__$synthetic$__%d", counter + 1), String::concat)
+				.withEvaluationContextProvider(evaluationContextProvider);
+
+		this.query = queryMethod.getRequiredQuery();
+		this.spelEvaluator = queryContext.parse(query, getQueryMethod().getParameters());
+		this.containsSpelExpressions = !this.spelEvaluator.getQueryString().equals(queryContext);
 	}
 
 	@Override
@@ -129,49 +151,38 @@ public class StringBasedJdbcQuery extends AbstractJdbcQuery {
 
 		RelationalParameterAccessor accessor = new RelationalParametersParameterAccessor(getQueryMethod(), objects);
 		ResultProcessor processor = getQueryMethod().getResultProcessor().withDynamicProjection(accessor);
-		ResultProcessingConverter converter = new ResultProcessingConverter(processor, this.converter.getMappingContext(),
-				this.converter.getEntityInstantiators());
-
-		JdbcQueryExecution<?> queryExecution = createJdbcQueryExecution(accessor, processor, converter);
 
+		JdbcQueryExecution<?> queryExecution = createJdbcQueryExecution(accessor, processor);
 		MapSqlParameterSource parameterMap = this.bindParameters(accessor);
 
-		String query = determineQuery();
+		return queryExecution.execute(processSpelExpressions(objects, parameterMap), parameterMap);
+	}
+
+	private String processSpelExpressions(Object[] objects, MapSqlParameterSource parameterMap) {
 
-		if (ObjectUtils.isEmpty(query)) {
-			throw new IllegalStateException(String.format("No query specified on %s", getQueryMethod().getName()));
+		if (containsSpelExpressions) {
+
+			spelEvaluator.evaluate(objects).forEach(parameterMap::addValue);
+			return spelEvaluator.getQueryString();
 		}
 
-		return queryExecution.execute(processSpelExpressions(objects, parameterMap, query), parameterMap);
+		return this.query;
 	}
 
 	private JdbcQueryExecution<?> createJdbcQueryExecution(RelationalParameterAccessor accessor,
-			ResultProcessor processor, ResultProcessingConverter converter) {
+			ResultProcessor processor) {
 
 		if (getQueryMethod().isModifyingQuery()) {
 			return createModifyingQueryExecutor();
 		} else {
 
-			RowMapper<Object> rowMapper = determineRowMapper(rowMapperFactory.create(resolveTypeToRead(processor)), converter,
-					accessor.findDynamicProjection() != null);
+			Supplier<RowMapper<?>> rowMapper = () -> determineRowMapper(processor, accessor.findDynamicProjection() != null);
+			ResultSetExtractor<Object> resultSetExtractor = determineResultSetExtractor(rowMapper);
 
-			return createReadingQueryExecution(determineResultSetExtractor(rowMapper), rowMapper);
+			return createReadingQueryExecution(resultSetExtractor, rowMapper);
 		}
 	}
 
-	private String processSpelExpressions(Object[] objects, MapSqlParameterSource parameterMap, String query) {
-
-		SpelQueryContext.EvaluatingSpelQueryContext queryContext = SpelQueryContext
-				.of((counter, expression) -> String.format("__$synthetic$__%d", counter + 1), String::concat)
-				.withEvaluationContextProvider(evaluationContextProvider);
-
-		SpelEvaluator spelEvaluator = queryContext.parse(query, getQueryMethod().getParameters());
-
-		spelEvaluator.evaluate(objects).forEach(parameterMap::addValue);
-
-		return spelEvaluator.getQueryString();
-	}
-
 	private MapSqlParameterSource bindParameters(RelationalParameterAccessor accessor) {
 
 		MapSqlParameterSource parameters = new MapSqlParameterSource();
@@ -189,7 +200,7 @@ public class StringBasedJdbcQuery extends AbstractJdbcQuery {
 
 		String parameterName = p.getName().orElseThrow(() -> new IllegalStateException(PARAMETER_NEEDS_TO_BE_NAMED));
 
-		RelationalParameters.RelationalParameter parameter = getQueryMethod().getParameters().getParameter(p.getIndex());
+		JdbcParameters.JdbcParameter parameter = getQueryMethod().getParameters().getParameter(p.getIndex());
 		TypeInformation<?> typeInformation = parameter.getTypeInformation();
 
 		JdbcValue jdbcValue;
@@ -200,8 +211,7 @@ public class StringBasedJdbcQuery extends AbstractJdbcQuery {
 
 			TypeInformation<?> actualType = typeInformation.getRequiredActualType();
 			for (Object o : (Iterable<?>) value) {
-				JdbcValue elementJdbcValue = converter.writeJdbcValue(o, actualType.getType(),
-						JdbcUtil.targetSqlTypeFor(JdbcColumnTypes.INSTANCE.resolvePrimitiveType(actualType.getType())));
+				JdbcValue elementJdbcValue = converter.writeJdbcValue(o, actualType, parameter.getActualSqlType());
 				if (jdbcType == null) {
 					jdbcType = elementJdbcValue.getJdbcType();
 				}
@@ -211,8 +221,8 @@ public class StringBasedJdbcQuery extends AbstractJdbcQuery {
 
 			jdbcValue = JdbcValue.of(mapped, jdbcType);
 		} else {
-			jdbcValue = converter.writeJdbcValue(value, typeInformation.getType(),
-					JdbcUtil.targetSqlTypeFor(JdbcColumnTypes.INSTANCE.resolvePrimitiveType(typeInformation.getType())));
+			SQLType sqlType = parameter.getSqlType();
+			jdbcValue = converter.writeJdbcValue(value, typeInformation, sqlType);
 		}
 
 		SQLType jdbcType = jdbcValue.getJdbcType();
@@ -224,86 +234,171 @@ public class StringBasedJdbcQuery extends AbstractJdbcQuery {
 		}
 	}
 
-	private String determineQuery() {
+	RowMapper<Object> determineRowMapper(ResultProcessor resultProcessor, boolean hasDynamicProjection) {
+
+		if (cachedRowMapperFactory.isConfiguredRowMapper()) {
+			return cachedRowMapperFactory.getRowMapper();
+		}
 
-		String query = getQueryMethod().getDeclaredQuery();
+		if (hasDynamicProjection) {
 
-		if (ObjectUtils.isEmpty(query)) {
-			throw new IllegalStateException(String.format("No query specified on %s", getQueryMethod().getName()));
+			RowMapper<Object> rowMapperToUse = rowMapperFactory.create(resultProcessor.getReturnedType().getDomainType());
+
+			ResultProcessingConverter converter = new ResultProcessingConverter(resultProcessor,
+					this.converter.getMappingContext(), this.converter.getEntityInstantiators());
+			return new ConvertingRowMapper<>(rowMapperToUse, converter);
 		}
 
-		return query;
+		return cachedRowMapperFactory.getRowMapper();
 	}
 
 	@Nullable
-	@SuppressWarnings({ "rawtypes", "unchecked" })
-	ResultSetExtractor<Object> determineResultSetExtractor(@Nullable RowMapper<Object> rowMapper) {
+	ResultSetExtractor<Object> determineResultSetExtractor(Supplier<RowMapper<?>> rowMapper) {
 
-		String resultSetExtractorRef = getQueryMethod().getResultSetExtractorRef();
+		if (cachedResultSetExtractorFactory.isConfiguredResultSetExtractor()) {
 
-		if (!ObjectUtils.isEmpty(resultSetExtractorRef)) {
-
-			Assert.notNull(beanFactory, "When a ResultSetExtractorRef is specified the BeanFactory must not be null");
+			if (cachedResultSetExtractorFactory.requiresRowMapper() && !cachedRowMapperFactory.isConfiguredRowMapper()) {
+				return cachedResultSetExtractorFactory.getResultSetExtractor(rowMapper);
+			}
 
-			return (ResultSetExtractor<Object>) beanFactory.getBean(resultSetExtractorRef);
+			// configured ResultSetExtractor defaults to configured RowMapper in case both are configured
+			return cachedResultSetExtractorFactory.getResultSetExtractor();
 		}
 
-		Class<? extends ResultSetExtractor> resultSetExtractorClass = getQueryMethod().getResultSetExtractorClass();
+		return null;
+	}
 
-		if (isUnconfigured(resultSetExtractorClass, ResultSetExtractor.class)) {
-			return null;
-		}
+	private static boolean isUnconfigured(@Nullable Class<?> configuredClass, Class<?> defaultClass) {
+		return configuredClass == null || configuredClass == defaultClass;
+	}
 
-		Constructor<? extends ResultSetExtractor> constructor = ClassUtils
-				.getConstructorIfAvailable(resultSetExtractorClass, RowMapper.class);
+	public void setBeanFactory(BeanFactory beanFactory) {
+		this.beanFactory = beanFactory;
+	}
 
-		if (constructor != null) {
-			return BeanUtils.instantiateClass(constructor, rowMapper);
-		}
+	class CachedRowMapperFactory {
 
-		return BeanUtils.instantiateClass(resultSetExtractorClass);
-	}
+		private final Lazy<RowMapper<Object>> cachedRowMapper;
+		private final boolean configuredRowMapper;
+		private final @Nullable Constructor<?> constructor;
 
-	@Nullable
-	RowMapper<Object> determineRowMapper(@Nullable RowMapper<?> defaultMapper,
-			Converter<Object, Object> resultProcessingConverter, boolean hasDynamicProjection) {
+		@SuppressWarnings("unchecked")
+		public CachedRowMapperFactory(Supplier<RowMapper<Object>> defaultMapper) {
 
-		RowMapper<Object> rowMapperToUse = determineRowMapper(defaultMapper);
+			String rowMapperRef = getQueryMethod().getRowMapperRef();
+			Class<?> rowMapperClass = getQueryMethod().getRowMapperClass();
 
-		if ((hasDynamicProjection || rowMapperToUse == defaultMapper) && rowMapperToUse != null) {
-			return new ConvertingRowMapper<>(rowMapperToUse, resultProcessingConverter);
+			if (!ObjectUtils.isEmpty(rowMapperRef) && !isUnconfigured(rowMapperClass, RowMapper.class)) {
+				throw new IllegalArgumentException(
+						"Invalid RowMapper configuration. Configure either one but not both via @Query(rowMapperRef = …, rowMapperClass = …) for query method "
+								+ getQueryMethod());
+			}
+
+			this.configuredRowMapper = !ObjectUtils.isEmpty(rowMapperRef) || !isUnconfigured(rowMapperClass, RowMapper.class);
+			this.constructor = rowMapperClass != null ? findPrimaryConstructor(rowMapperClass) : null;
+			this.cachedRowMapper = Lazy.of(() -> {
+
+				if (!ObjectUtils.isEmpty(rowMapperRef)) {
+
+					Assert.notNull(beanFactory, "When a RowMapperRef is specified the BeanFactory must not be null");
+
+					return (RowMapper<Object>) beanFactory.getBean(rowMapperRef);
+				}
+
+				if (isUnconfigured(rowMapperClass, RowMapper.class)) {
+					return defaultMapper.get();
+				}
+
+				return (RowMapper<Object>) BeanUtils.instantiateClass(constructor);
+			});
 		}
 
-		return rowMapperToUse;
+		public boolean isConfiguredRowMapper() {
+			return configuredRowMapper;
+		}
+
+		public RowMapper<Object> getRowMapper() {
+			return cachedRowMapper.get();
+		}
 	}
 
-	@SuppressWarnings("unchecked")
-	@Nullable
-	RowMapper<Object> determineRowMapper(@Nullable RowMapper<?> defaultMapper) {
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	class CachedResultSetExtractorFactory {
+
+		private final Lazy<ResultSetExtractor<Object>> cachedResultSetExtractor;
+		private final boolean configuredResultSetExtractor;
+		private final @Nullable Constructor<? extends ResultSetExtractor> rowMapperConstructor;
+		private final @Nullable Constructor<? extends ResultSetExtractor> constructor;
+		private final Function<Supplier<RowMapper<?>>, ResultSetExtractor<Object>> resultSetExtractorFactory;
+
+		public CachedResultSetExtractorFactory(Supplier<RowMapper<?>> resultSetExtractor) {
+
+			String resultSetExtractorRef = getQueryMethod().getResultSetExtractorRef();
+			Class<? extends ResultSetExtractor> resultSetExtractorClass = getQueryMethod().getResultSetExtractorClass();
+
+			if (!ObjectUtils.isEmpty(resultSetExtractorRef)
+					&& !isUnconfigured(resultSetExtractorClass, ResultSetExtractor.class)) {
+				throw new IllegalArgumentException(
+						"Invalid ResultSetExtractor configuration. Configure either one but not both via @Query(resultSetExtractorRef = …, resultSetExtractorClass = …) for query method "
+								+ getQueryMethod());
+			}
 
-		String rowMapperRef = getQueryMethod().getRowMapperRef();
+			this.configuredResultSetExtractor = !ObjectUtils.isEmpty(resultSetExtractorRef)
+					|| !isUnconfigured(resultSetExtractorClass, ResultSetExtractor.class);
 
-		if (!ObjectUtils.isEmpty(rowMapperRef)) {
+			this.rowMapperConstructor = resultSetExtractorClass != null
+					? ClassUtils.getConstructorIfAvailable(resultSetExtractorClass, RowMapper.class)
+					: null;
+			this.constructor = resultSetExtractorClass != null ? findPrimaryConstructor(resultSetExtractorClass) : null;
+			this.resultSetExtractorFactory = rowMapper -> {
 
-			Assert.notNull(beanFactory, "When a RowMapperRef is specified the BeanFactory must not be null");
+				if (!ObjectUtils.isEmpty(resultSetExtractorRef)) {
 
-			return (RowMapper<Object>) beanFactory.getBean(rowMapperRef);
+					Assert.notNull(beanFactory, "When a ResultSetExtractorRef is specified the BeanFactory must not be null");
+
+					return (ResultSetExtractor<Object>) beanFactory.getBean(resultSetExtractorRef);
+				}
+
+				if (isUnconfigured(resultSetExtractorClass, ResultSetExtractor.class)) {
+					throw new UnsupportedOperationException("This should not happen");
+				}
+
+				if (rowMapperConstructor != null) {
+					return BeanUtils.instantiateClass(rowMapperConstructor, rowMapper.get());
+				}
+
+				return BeanUtils.instantiateClass(constructor);
+			};
+
+			this.cachedResultSetExtractor = Lazy.of(() -> resultSetExtractorFactory.apply(resultSetExtractor));
 		}
 
-		Class<?> rowMapperClass = getQueryMethod().getRowMapperClass();
+		public boolean isConfiguredResultSetExtractor() {
+			return configuredResultSetExtractor;
+		}
 
-		if (isUnconfigured(rowMapperClass, RowMapper.class)) {
-			return (RowMapper<Object>) defaultMapper;
+		public ResultSetExtractor<Object> getResultSetExtractor() {
+			return cachedResultSetExtractor.get();
 		}
 
-		return (RowMapper<Object>) BeanUtils.instantiateClass(rowMapperClass);
-	}
+		public ResultSetExtractor<Object> getResultSetExtractor(Supplier<RowMapper<?>> rowMapperSupplier) {
+			return resultSetExtractorFactory.apply(rowMapperSupplier);
+		}
 
-	private static boolean isUnconfigured(@Nullable Class<?> configuredClass, Class<?> defaultClass) {
-		return configuredClass == null || configuredClass == defaultClass;
+		public boolean requiresRowMapper() {
+			return rowMapperConstructor != null;
+		}
 	}
 
-	public void setBeanFactory(BeanFactory beanFactory) {
-		this.beanFactory = beanFactory;
+	@Nullable
+	static <T> Constructor<T> findPrimaryConstructor(Class<T> clazz) {
+		try {
+			return clazz.getDeclaredConstructor();
+		} catch (NoSuchMethodException ex) {
+			return BeanUtils.findPrimaryConstructor(clazz);
+
+		} catch (LinkageError err) {
+			throw new BeanInstantiationException(clazz, "Unresolvable class definition", err);
+		}
 	}
 }

spring-data-jdbc/src/test/java/org/springframework/data/jdbc/repository/query/StringBasedJdbcQueryUnitTests.java

@@ -32,6 +32,8 @@ import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
+
+import org.springframework.beans.factory.BeanFactory;
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.dao.DataAccessException;
 import org.springframework.data.convert.ReadingConverter;
@@ -59,6 +61,7 @@ import org.springframework.jdbc.core.ResultSetExtractor;
 import org.springframework.jdbc.core.RowMapper;
 import org.springframework.jdbc.core.namedparam.NamedParameterJdbcOperations;
 import org.springframework.jdbc.core.namedparam.SqlParameterSource;
+import org.springframework.test.util.ReflectionTestUtils;
 import org.springframework.util.ReflectionUtils;
 
 /**
@@ -106,7 +109,7 @@ class StringBasedJdbcQueryUnitTests {
 		JdbcQueryMethod queryMethod = createMethod("findAll");
 		StringBasedJdbcQuery query = createQuery(queryMethod);
 
-		assertThat(query.determineRowMapper(defaultRowMapper)).isEqualTo(defaultRowMapper);
+		assertThat(query.determineRowMapper(queryMethod.getResultProcessor(), false)).isEqualTo(defaultRowMapper);
 	}
 
 	@Test // DATAJDBC-165, DATAJDBC-318
@@ -115,7 +118,7 @@ class StringBasedJdbcQueryUnitTests {
 		JdbcQueryMethod queryMethod = createMethod("findAllWithCustomRowMapper");
 		StringBasedJdbcQuery query = createQuery(queryMethod);
 
-		assertThat(query.determineRowMapper(defaultRowMapper)).isInstanceOf(CustomRowMapper.class);
+		assertThat(query.determineRowMapper(queryMethod.getResultProcessor(), false)).isInstanceOf(CustomRowMapper.class);
 	}
 
 	@Test // DATAJDBC-290
@@ -124,12 +127,90 @@ class StringBasedJdbcQueryUnitTests {
 		JdbcQueryMethod queryMethod = createMethod("findAllWithCustomResultSetExtractor");
 		StringBasedJdbcQuery query = createQuery(queryMethod);
 
-		ResultSetExtractor<Object> resultSetExtractor = query.determineResultSetExtractor(defaultRowMapper);
+		ResultSetExtractor<Object> resultSetExtractor1 = query.determineResultSetExtractor(() -> defaultRowMapper);
+		ResultSetExtractor<Object> resultSetExtractor2 = query.determineResultSetExtractor(() -> defaultRowMapper);
 
-		assertThat(resultSetExtractor) //
+		assertThat(resultSetExtractor1) //
 				.isInstanceOf(CustomResultSetExtractor.class) //
 				.matches(crse -> ((CustomResultSetExtractor) crse).rowMapper == defaultRowMapper,
 						"RowMapper is expected to be default.");
+
+		assertThat(resultSetExtractor1).isNotSameAs(resultSetExtractor2);
+	}
+
+	@Test // GH-1721
+	void cachesCustomMapperAndExtractorInstances() {
+
+		JdbcQueryMethod queryMethod = createMethod("findAllCustomRowMapperResultSetExtractor");
+		StringBasedJdbcQuery query = createQuery(queryMethod);
+
+		ResultSetExtractor<Object> resultSetExtractor1 = query.determineResultSetExtractor(() -> {
+			throw new UnsupportedOperationException();
+		});
+
+		ResultSetExtractor<Object> resultSetExtractor2 = query.determineResultSetExtractor(() -> {
+			throw new UnsupportedOperationException();
+		});
+
+		assertThat(resultSetExtractor1).isSameAs(resultSetExtractor2);
+		assertThat(resultSetExtractor1).extracting("rowMapper").isInstanceOf(CustomRowMapper.class);
+
+		assertThat(ReflectionTestUtils.getField(resultSetExtractor1, "rowMapper"))
+				.isSameAs(ReflectionTestUtils.getField(resultSetExtractor2, "rowMapper"));
+	}
+
+	@Test // GH-1721
+	void obtainsCustomRowMapperRef() {
+
+		BeanFactory beanFactory = mock(BeanFactory.class);
+		JdbcQueryMethod queryMethod = createMethod("findAllCustomRowMapperRef");
+		StringBasedJdbcQuery query = createQuery(queryMethod);
+		query.setBeanFactory(beanFactory);
+
+		CustomRowMapper customRowMapper = new CustomRowMapper();
+
+		when(beanFactory.getBean("CustomRowMapper")).thenReturn(customRowMapper);
+
+		RowMapper<?> rowMapper = query.determineRowMapper(queryMethod.getResultProcessor(), false);
+		ResultSetExtractor<Object> resultSetExtractor = query.determineResultSetExtractor(() -> {
+			throw new UnsupportedOperationException();
+		});
+
+		assertThat(rowMapper).isSameAs(customRowMapper);
+		assertThat(resultSetExtractor).isNull();
+	}
+
+	@Test // GH-1721
+	void obtainsCustomResultSetExtractorRef() {
+
+		BeanFactory beanFactory = mock(BeanFactory.class);
+		JdbcQueryMethod queryMethod = createMethod("findAllCustomResultSetExtractorRef");
+		StringBasedJdbcQuery query = createQuery(queryMethod);
+		query.setBeanFactory(beanFactory);
+
+		CustomResultSetExtractor cre = new CustomResultSetExtractor();
+
+		when(beanFactory.getBean("CustomResultSetExtractor")).thenReturn(cre);
+
+		RowMapper<?> rowMapper = query.determineRowMapper(queryMethod.getResultProcessor(), false);
+		ResultSetExtractor<Object> resultSetExtractor = query.determineResultSetExtractor(() -> {
+			throw new UnsupportedOperationException();
+		});
+
+		assertThat(rowMapper).isSameAs(defaultRowMapper);
+		assertThat(resultSetExtractor).isSameAs(cre);
+	}
+
+	@Test // GH-1721
+	void failsOnRowMapperRefAndClassDeclaration() {
+		assertThatIllegalArgumentException().isThrownBy(() -> createQuery(createMethod("invalidMapperRefAndClass")))
+				.withMessageContaining("Invalid RowMapper configuration");
+	}
+
+	@Test // GH-1721
+	void failsOnResultSetExtractorRefAndClassDeclaration() {
+		assertThatIllegalArgumentException().isThrownBy(() -> createQuery(createMethod("invalidExtractorRefAndClass")))
+				.withMessageContaining("Invalid ResultSetExtractor configuration");
 	}
 
 	@Test // DATAJDBC-290
@@ -139,7 +220,7 @@ class StringBasedJdbcQueryUnitTests {
 		StringBasedJdbcQuery query = createQuery(queryMethod);
 
 		ResultSetExtractor<Object> resultSetExtractor = query
-				.determineResultSetExtractor(query.determineRowMapper(defaultRowMapper));
+				.determineResultSetExtractor(() -> query.determineRowMapper(queryMethod.getResultProcessor(), false));
 
 		assertThat(resultSetExtractor) //
 				.isInstanceOf(CustomResultSetExtractor.class) //
@@ -178,8 +259,8 @@ class StringBasedJdbcQueryUnitTests {
 
 		assertThatThrownBy(
 				() -> new StringBasedJdbcQuery(queryMethod, operations, defaultRowMapper, converter, evaluationContextProvider))
-						.isInstanceOf(UnsupportedOperationException.class)
-						.hasMessageContaining("Slice queries are not supported using string-based queries");
+				.isInstanceOf(UnsupportedOperationException.class)
+				.hasMessageContaining("Slice queries are not supported using string-based queries");
 	}
 
 	@Test // GH-774
@@ -189,8 +270,8 @@ class StringBasedJdbcQueryUnitTests {
 
 		assertThatThrownBy(
 				() -> new StringBasedJdbcQuery(queryMethod, operations, defaultRowMapper, converter, evaluationContextProvider))
-						.isInstanceOf(UnsupportedOperationException.class)
-						.hasMessageContaining("Page queries are not supported using string-based queries");
+				.isInstanceOf(UnsupportedOperationException.class)
+				.hasMessageContaining("Page queries are not supported using string-based queries");
 	}
 
 	@Test // GH-1654
@@ -200,7 +281,7 @@ class StringBasedJdbcQueryUnitTests {
 
 		assertThatThrownBy(
 				() -> new StringBasedJdbcQuery(queryMethod, operations, defaultRowMapper, converter, evaluationContextProvider))
-						.isInstanceOf(UnsupportedOperationException.class);
+				.isInstanceOf(UnsupportedOperationException.class);
 	}
 
 	@Test // GH-1212
@@ -329,6 +410,24 @@ class StringBasedJdbcQueryUnitTests {
 		@Query(value = "some sql statement", resultSetExtractorClass = CustomResultSetExtractor.class)
 		Stream<Object> findAllWithStreamReturnTypeAndResultSetExtractor();
 
+		@Query(value = "some sql statement", rowMapperClass = CustomRowMapper.class,
+				resultSetExtractorClass = CustomResultSetExtractor.class)
+		Stream<Object> findAllCustomRowMapperResultSetExtractor();
+
+		@Query(value = "some sql statement", rowMapperRef = "CustomRowMapper")
+		Stream<Object> findAllCustomRowMapperRef();
+
+		@Query(value = "some sql statement", resultSetExtractorRef = "CustomResultSetExtractor")
+		Stream<Object> findAllCustomResultSetExtractorRef();
+
+		@Query(value = "some sql statement", rowMapperRef = "CustomResultSetExtractor",
+				rowMapperClass = CustomRowMapper.class)
+		Stream<Object> invalidMapperRefAndClass();
+
+		@Query(value = "some sql statement", resultSetExtractorRef = "CustomResultSetExtractor",
+				resultSetExtractorClass = CustomResultSetExtractor.class)
+		Stream<Object> invalidExtractorRefAndClass();
+
 		List<Object> noAnnotation();
 
 		@Query(value = "some sql statement")

spring-data-relational/src/main/java/org/springframework/data/relational/core/conversion/MappingRelationalConverter.java

@@ -77,14 +77,15 @@ import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
 
 /**
- * {@link RelationalConverter} that uses a {@link MappingContext} to apply sophisticated mapping of domain objects from
- * {@link RowDocument}.
+ * {@link org.springframework.data.relational.core.conversion.RelationalConverter} that uses a
+ * {@link org.springframework.data.mapping.context.MappingContext} to apply sophisticated mapping of domain objects from
+ * {@link org.springframework.data.relational.domain.RowDocument}.
  *
  * @author Mark Paluch
  * @author Jens Schauder
  * @author Chirag Tailor
  * @author Vincent Galloy
- * @see MappingContext
+ * @see org.springframework.data.mapping.context.MappingContext
  * @see SimpleTypeHolder
  * @see CustomConversions
  * @since 3.2
@@ -701,8 +702,25 @@ public class MappingRelationalConverter extends AbstractRelationalConverter
 
 		if (getConversions().isSimpleType(value.getClass())) {
 
-			if (TypeInformation.OBJECT != type && getConversionService().canConvert(value.getClass(), type.getType())) {
-				value = getConversionService().convert(value, type.getType());
+			Optional<Class<?>> customWriteTarget = getConversions().getCustomWriteTarget(type.getType());
+			if (customWriteTarget.isPresent()) {
+				return getConversionService().convert(value, customWriteTarget.get());
+			}
+
+			if (TypeInformation.OBJECT != type) {
+
+				if (type.getType().isAssignableFrom(value.getClass())) {
+
+					if (value.getClass().isEnum()) {
+						return getPotentiallyConvertedSimpleWrite(value);
+					}
+
+					return value;
+				} else {
+					if (getConversionService().canConvert(value.getClass(), type.getType())) {
+						value = getConversionService().convert(value, type.getType());
+					}
+				}
 			}
 
 			return getPotentiallyConvertedSimpleWrite(value);
@@ -724,7 +742,9 @@ public class MappingRelationalConverter extends AbstractRelationalConverter
 			return writeValue(id, type);
 		}
 
-		return getConversionService().convert(value, type.getType());
+		return
+
+		getConversionService().convert(value, type.getType());
 	}
 
 	private Object writeArray(Object value, TypeInformation<?> type) {
@@ -1207,8 +1227,7 @@ public class MappingRelationalConverter extends AbstractRelationalConverter
 		 * @param delegate must not be {@literal null}.
 		 */
 		public ConverterAwareExpressionParameterValueProvider(ConversionContext context, ValueExpressionEvaluator evaluator,
-				ConversionService conversionService,
-				ParameterValueProvider<RelationalPersistentProperty> delegate) {
+				ConversionService conversionService, ParameterValueProvider<RelationalPersistentProperty> delegate) {
 
 			super(evaluator, conversionService, delegate);
 

spring-data-relational/src/main/java/org/springframework/data/relational/core/mapping/DefaultAggregatePath.java

@@ -23,6 +23,7 @@ import org.springframework.data.mapping.PersistentPropertyPath;
 import org.springframework.data.util.Lazy;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
+import org.springframework.util.ConcurrentLruCache;
 
 /**
  * Represents a path within an aggregate starting from the aggregate root.
@@ -43,6 +44,8 @@ class DefaultAggregatePath implements AggregatePath {
 
 	private final Lazy<ColumnInfo> columnInfo = Lazy.of(() -> ColumnInfo.of(this));
 
+	private final ConcurrentLruCache<RelationalPersistentProperty, AggregatePath> nestedCache;
+
 	@SuppressWarnings("unchecked")
 	DefaultAggregatePath(RelationalMappingContext context,
 			PersistentPropertyPath<? extends RelationalPersistentProperty> path) {
@@ -53,6 +56,7 @@ class DefaultAggregatePath implements AggregatePath {
 		this.context = context;
 		this.path = (PersistentPropertyPath) path;
 		this.rootType = path.getBaseProperty().getOwner();
+		this.nestedCache = new ConcurrentLruCache<>(32, this::doGetAggegatePath);
 	}
 
 	DefaultAggregatePath(RelationalMappingContext context, RelationalPersistentEntity<?> rootType) {
@@ -63,6 +67,7 @@ class DefaultAggregatePath implements AggregatePath {
 		this.context = context;
 		this.rootType = rootType;
 		this.path = null;
+		this.nestedCache = new ConcurrentLruCache<>(32, this::doGetAggegatePath);
 	}
 
 	/**
@@ -89,6 +94,10 @@ class DefaultAggregatePath implements AggregatePath {
 
 	@Override
 	public AggregatePath append(RelationalPersistentProperty property) {
+		return nestedCache.get(property);
+	}
+
+	private AggregatePath doGetAggegatePath(RelationalPersistentProperty property) {
 
 		PersistentPropertyPath<? extends RelationalPersistentProperty> newPath = isRoot() //
 				? context.getPersistentPropertyPath(property.getName(), rootType.getTypeInformation()) //

spring-data-relational/src/main/java/org/springframework/data/relational/core/mapping/DerivedSqlIdentifier.java

@@ -36,12 +36,15 @@ class DerivedSqlIdentifier implements SqlIdentifier {
 
 	private final String name;
 	private final boolean quoted;
+	private final String toString;
+	private volatile @Nullable CachedSqlName sqlName;
 
 	DerivedSqlIdentifier(String name, boolean quoted) {
 
 		Assert.hasText(name, "A database object must have at least on name part.");
 		this.name = name;
 		this.quoted = quoted;
+		this.toString = quoted ? toSql(IdentifierProcessing.ANSI) : this.name;
 	}
 
 	@Override
@@ -60,13 +63,19 @@ class DerivedSqlIdentifier implements SqlIdentifier {
 	@Override
 	public String toSql(IdentifierProcessing processing) {
 
-		String normalized = processing.standardizeLetterCase(name);
+		CachedSqlName sqlName = this.sqlName;
+		if (sqlName == null || sqlName.processing != processing) {
 
-		return quoted ? processing.quote(normalized) : normalized;
+			String normalized = processing.standardizeLetterCase(name);
+			this.sqlName = sqlName = new CachedSqlName(processing, quoted ? processing.quote(normalized) : normalized);
+			return sqlName.sqlName();
+		}
+
+		return sqlName.sqlName();
 	}
 
 	@Override
-	@Deprecated(since="3.1", forRemoval = true)
+	@Deprecated(since = "3.1", forRemoval = true)
 	public String getReference(IdentifierProcessing processing) {
 		return this.name;
 	}
@@ -92,6 +101,9 @@ class DerivedSqlIdentifier implements SqlIdentifier {
 
 	@Override
 	public String toString() {
-		return quoted ? toSql(IdentifierProcessing.ANSI) : this.name;
+		return toString;
+	}
+
+	record CachedSqlName(IdentifierProcessing processing, String sqlName) {
 	}
 }

spring-data-relational/src/main/java/org/springframework/data/relational/core/sql/DefaultSqlIdentifier.java

@@ -34,6 +34,8 @@ class DefaultSqlIdentifier implements SqlIdentifier {
 
 	private final String name;
 	private final boolean quoted;
+	private final String toString;
+	private volatile @Nullable CachedSqlName sqlName;
 
 	DefaultSqlIdentifier(String name, boolean quoted) {
 
@@ -41,6 +43,7 @@ class DefaultSqlIdentifier implements SqlIdentifier {
 
 		this.name = name;
 		this.quoted = quoted;
+		this.toString = quoted ? toSql(IdentifierProcessing.ANSI) : this.name;
 	}
 
 	@Override
@@ -58,11 +61,19 @@ class DefaultSqlIdentifier implements SqlIdentifier {
 
 	@Override
 	public String toSql(IdentifierProcessing processing) {
-		return quoted ? processing.quote(name) : name;
+
+		CachedSqlName sqlName = this.sqlName;
+		if (sqlName == null || sqlName.processing != processing) {
+
+			this.sqlName = sqlName = new CachedSqlName(processing, quoted ? processing.quote(name) : name);
+			return sqlName.sqlName();
+		}
+
+		return sqlName.sqlName();
 	}
 
 	@Override
-	@Deprecated(since="3.1", forRemoval = true)
+	@Deprecated(since = "3.1", forRemoval = true)
 	public String getReference(IdentifierProcessing processing) {
 		return name;
 	}
@@ -88,6 +99,9 @@ class DefaultSqlIdentifier implements SqlIdentifier {
 
 	@Override
 	public String toString() {
-		return quoted ? toSql(IdentifierProcessing.ANSI) : this.name;
+		return toString;
+	}
+
+	record CachedSqlName(IdentifierProcessing processing, String sqlName) {
 	}
 }

spring-data-relational/src/main/java/org/springframework/data/relational/repository/query/RelationalParameters.java

@@ -16,6 +16,7 @@
 package org.springframework.data.relational.repository.query;
 
 import java.util.List;
+import java.util.function.Function;
 
 import org.springframework.core.MethodParameter;
 import org.springframework.core.ResolvableType;
@@ -42,7 +43,12 @@ public class RelationalParameters extends Parameters<RelationalParameters, Relat
 				methodParameter -> new RelationalParameter(methodParameter, parametersSource.getDomainTypeInformation()));
 	}
 
-	private RelationalParameters(List<RelationalParameter> parameters) {
+	protected RelationalParameters(ParametersSource parametersSource,
+			Function<MethodParameter, RelationalParameter> parameterFactory) {
+		super(parametersSource, parameterFactory);
+	}
+
+	protected RelationalParameters(List<RelationalParameter> parameters) {
 		super(parameters);
 	}
 
@@ -59,16 +65,17 @@ public class RelationalParameters extends Parameters<RelationalParameters, Relat
 	 */
 	public static class RelationalParameter extends Parameter {
 
-		private final MethodParameter parameter;
+		private final TypeInformation<?> typeInformation;
 
 		/**
 		 * Creates a new {@link RelationalParameter}.
 		 *
 		 * @param parameter must not be {@literal null}.
 		 */
-		RelationalParameter(MethodParameter parameter, TypeInformation<?> domainType) {
+		protected RelationalParameter(MethodParameter parameter, TypeInformation<?> domainType) {
 			super(parameter, domainType);
-			this.parameter = parameter;
+			this.typeInformation = TypeInformation.fromMethodParameter(parameter);
+
 		}
 
 		public ResolvableType getResolvableType() {
@@ -76,7 +83,7 @@ public class RelationalParameters extends Parameters<RelationalParameters, Relat
 		}
 
 		public TypeInformation<?> getTypeInformation() {
-			return TypeInformation.fromMethodParameter(parameter);
+			return typeInformation;
 		}
 	}
 }

spring-data-relational/src/test/java/org/springframework/data/relational/core/conversion/BasicRelationalConverterUnitTests.java

@@ -133,7 +133,7 @@ class BasicRelationalConverterUnitTests {
 	@Test // DATAJDBC-516
 	void shouldConsiderWriteConverter() {
 
-		Object result = converter.writeValue(new MyValue("hello-world"), TypeInformation.of(MyValue.class));
+		Object result = converter.writeValue(new MyValue("hello-world"), TypeInformation.of(String.class));
 
 		assertThat(result).isEqualTo("hello-world");
 	}