Introduce JdbcClient as a fluent facade for query/update execution

Delegates to JdbcTemplate/NamedParameterJdbcTemplate underneath the covers.
Supports parameter objects/records through SimplePropertySqlParameterSource.

Closes gh-30931

spring-jdbc/src/main/java/org/springframework/jdbc/core/JdbcOperations.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2021 the original author or authors.
+ * Copyright 2002-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.
@@ -35,12 +35,19 @@ import org.springframework.lang.Nullable;
  * <p>Alternatively, the standard JDBC infrastructure can be mocked.
  * However, mocking this interface constitutes significantly less work.
  * As an alternative to a mock objects approach to testing data access code,
- * consider the powerful integration testing support provided via the <em>Spring
- * TestContext Framework</em>, in the {@code spring-test} artifact.
+ * consider the powerful integration testing support provided via the
+ * <em>Spring TestContext Framework</em>, in the {@code spring-test} artifact.
+ *
+ * <p><b>NOTE: As of 6.1, there is a unified JDBC access facade available in
+ * the form of {@link org.springframework.jdbc.core.simple.JdbcClient}.</b>
+ * {@code JdbcClient} provides a fluent API style for common JDBC queries/updates
+ * with flexible use of indexed or named parameters. It delegates to
+ * {@code JdbcOperations}/{@code NamedParameterJdbcOperations} for actual execution.
  *
  * @author Rod Johnson
  * @author Juergen Hoeller
  * @see JdbcTemplate
+ * @see org.springframework.jdbc.core.namedparam.NamedParameterJdbcOperations
  */
 public interface JdbcOperations {
 

spring-jdbc/src/main/java/org/springframework/jdbc/core/JdbcTemplate.java

@@ -65,8 +65,7 @@ import org.springframework.util.StringUtils;
  * It executes core JDBC workflow, leaving application code to provide SQL
  * and extract results. This class executes SQL queries or updates, initiating
  * iteration over ResultSets and catching JDBC exceptions and translating
- * them to the generic, more informative exception hierarchy defined in the
- * {@code org.springframework.dao} package.
+ * them to the common {@code org.springframework.dao} exception hierarchy.
  *
  * <p>Code using this class need only implement callback interfaces, giving
  * them a clearly defined contract. The {@link PreparedStatementCreator} callback
@@ -75,7 +74,8 @@ import org.springframework.util.StringUtils;
  * values from a ResultSet. See also {@link PreparedStatementSetter} and
  * {@link RowMapper} for two popular alternative callback interfaces.
  *
- * <p>Can be used within a service implementation via direct instantiation
+ * <p>An instance of this template class is thread-safe once configured.
+ * Can be used within a service implementation via direct instantiation
  * with a DataSource reference, or get prepared in an application context
  * and given to services as bean reference. Note: The DataSource should
  * always be configured as a bean in the application context, in the first case
@@ -88,12 +88,17 @@ import org.springframework.util.StringUtils;
  * <p>All SQL operations performed by this class are logged at debug level,
  * using "org.springframework.jdbc.core.JdbcTemplate" as log category.
  *
- * <p><b>NOTE: An instance of this class is thread-safe once configured.</b>
+ * <p><b>NOTE: As of 6.1, there is a unified JDBC access facade available in
+ * the form of {@link org.springframework.jdbc.core.simple.JdbcClient}.</b>
+ * {@code JdbcClient} provides a fluent API style for common JDBC queries/updates
+ * with flexible use of indexed or named parameters. It delegates to a
+ * {@code JdbcTemplate}/{@code NamedParameterJdbcTemplate} for actual execution.
  *
  * @author Rod Johnson
  * @author Juergen Hoeller
  * @author Thomas Risberg
  * @since May 3, 2001
+ * @see JdbcOperations
  * @see PreparedStatementCreator
  * @see PreparedStatementSetter
  * @see CallableStatementCreator
@@ -103,6 +108,7 @@ import org.springframework.util.StringUtils;
  * @see RowCallbackHandler
  * @see RowMapper
  * @see org.springframework.jdbc.support.SQLExceptionTranslator
+ * @see org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate
  */
 public class JdbcTemplate extends JdbcAccessor implements JdbcOperations {
 

spring-jdbc/src/main/java/org/springframework/jdbc/core/namedparam/BeanPropertySqlParameterSource.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-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.
@@ -32,15 +32,16 @@ import org.springframework.util.StringUtils;
 /**
  * {@link SqlParameterSource} implementation that obtains parameter values
  * from bean properties of a given JavaBean object. The names of the bean
- * properties have to match the parameter names.
+ * properties have to match the parameter names. Supports components of
+ * record classes as well, with accessor methods matching parameter names.
  *
- * <p>Uses a Spring BeanWrapper for bean property access underneath.
+ * <p>Uses a Spring {@link BeanWrapper} for bean property access underneath.
  *
  * @author Thomas Risberg
  * @author Juergen Hoeller
  * @since 2.0
  * @see NamedParameterJdbcTemplate
- * @see org.springframework.beans.BeanWrapper
+ * @see SimplePropertySqlParameterSource
  */
 public class BeanPropertySqlParameterSource extends AbstractSqlParameterSource {
 

spring-jdbc/src/main/java/org/springframework/jdbc/core/namedparam/MapSqlParameterSource.java

@@ -143,6 +143,14 @@ public class MapSqlParameterSource extends AbstractSqlParameterSource {
 		return this;
 	}
 
+	/**
+	 * Return whether this parameter source has been configured with any values.
+	 * @since 6.1
+	 */
+	public boolean hasValues() {
+		return !this.values.isEmpty();
+	}
+
 	/**
 	 * Expose the current parameter values as read-only Map.
 	 */

spring-jdbc/src/main/java/org/springframework/jdbc/core/namedparam/NamedParameterJdbcOperations.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2021 the original author or authors.
+ * Copyright 2002-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.
@@ -40,6 +40,12 @@ import org.springframework.lang.Nullable;
  * often used directly, but provides a useful option to enhance testability,
  * as it can easily be mocked or stubbed.
  *
+ * <p><b>NOTE: As of 6.1, there is a unified JDBC access facade available in
+ * the form of {@link org.springframework.jdbc.core.simple.JdbcClient}.</b>
+ * {@code JdbcClient} provides a fluent API style for common JDBC queries/updates
+ * with flexible use of indexed or named parameters. It delegates to
+ * {@code JdbcOperations}/{@code NamedParameterJdbcOperations} for actual execution.
+ *
  * @author Thomas Risberg
  * @author Juergen Hoeller
  * @since 2.0

spring-jdbc/src/main/java/org/springframework/jdbc/core/namedparam/NamedParameterJdbcTemplate.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-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.
@@ -55,16 +55,25 @@ import org.springframework.util.ConcurrentLruCache;
  * done at execution time. It also allows for expanding a {@link java.util.List}
  * of values to the appropriate number of placeholders.
  *
- * <p>The underlying {@link org.springframework.jdbc.core.JdbcTemplate} is
+ * <p>An instance of this template class is thread-safe once configured.
+ * The underlying {@link org.springframework.jdbc.core.JdbcTemplate} is
  * exposed to allow for convenient access to the traditional
  * {@link org.springframework.jdbc.core.JdbcTemplate} methods.
  *
- * <p><b>NOTE: An instance of this class is thread-safe once configured.</b>
+ * <p><b>NOTE: As of 6.1, there is a unified JDBC access facade available in
+ * the form of {@link org.springframework.jdbc.core.simple.JdbcClient}.</b>
+ * {@code JdbcClient} provides a fluent API style for common JDBC queries/updates
+ * with flexible use of indexed or named parameters. It delegates to a
+ * {@code JdbcTemplate}/{@code NamedParameterJdbcTemplate} for actual execution.
  *
  * @author Thomas Risberg
  * @author Juergen Hoeller
  * @since 2.0
  * @see NamedParameterJdbcOperations
+ * @see SqlParameterSource
+ * @see ResultSetExtractor
+ * @see RowCallbackHandler
+ * @see RowMapper
  * @see org.springframework.jdbc.core.JdbcTemplate
  */
 public class NamedParameterJdbcTemplate implements NamedParameterJdbcOperations {

spring-jdbc/src/main/java/org/springframework/jdbc/core/namedparam/SimplePropertySqlParameterSource.java

@@ -0,0 +1,117 @@
+/*
+ * Copyright 2002-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.jdbc.core.namedparam;
+
+import java.beans.PropertyDescriptor;
+import java.lang.reflect.Field;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.springframework.beans.BeanUtils;
+import org.springframework.jdbc.core.StatementCreatorUtils;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+import org.springframework.util.ReflectionUtils;
+
+/**
+ * {@link SqlParameterSource} implementation that obtains parameter values
+ * from bean properties of a given JavaBean object, from component accessors
+ * of a record class, or from raw field access.
+ *
+ * <p>This is a more flexible variant of {@link BeanPropertySqlParameterSource},
+ * with the limitation that it is not able to enumerate its
+ * {@link #getParameterNames() parameter names}.
+ *
+ * <p>In terms of its fallback property discovery algorithm, this class is
+ * similar to {@link org.springframework.validation.SimpleErrors} which is
+ * also just used for property retrieval purposes (rather than binding).
+ *
+ * @author Juergen Hoeller
+ * @since 6.1
+ * @see NamedParameterJdbcTemplate
+ * @see BeanPropertySqlParameterSource
+ */
+public class SimplePropertySqlParameterSource extends AbstractSqlParameterSource {
+
+	private final Object paramObject;
+
+	private final Map<String, Object> descriptorMap = new HashMap<>();
+
+
+	/**
+	 * Create a new SqlParameterSource for the given bean, record or field holder.
+	 * @param paramObject the bean, record or field holder instance to wrap
+	 */
+	public SimplePropertySqlParameterSource(Object paramObject) {
+		Assert.notNull(paramObject, "Parameter object must not be null");
+		this.paramObject = paramObject;
+	}
+
+
+	@Override
+	public boolean hasValue(String paramName) {
+		return (getDescriptor(paramName) != null);
+	}
+
+	@Override
+	@Nullable
+	public Object getValue(String paramName) throws IllegalArgumentException {
+		Object desc = getDescriptor(paramName);
+		if (desc instanceof PropertyDescriptor pd) {
+			ReflectionUtils.makeAccessible(pd.getReadMethod());
+			return ReflectionUtils.invokeMethod(pd.getReadMethod(), this.paramObject);
+		}
+		else if (desc instanceof Field field) {
+			ReflectionUtils.makeAccessible(field);
+			return ReflectionUtils.getField(field, this.paramObject);
+		}
+		throw new IllegalArgumentException("Cannot retrieve value for parameter '" + paramName +
+				"' - neither a getter method nor a raw field found");
+	}
+
+	/**
+	 * Derives a default SQL type from the corresponding property type.
+	 * @see StatementCreatorUtils#javaTypeToSqlParameterType
+	 */
+	@Override
+	public int getSqlType(String paramName) {
+		int sqlType = super.getSqlType(paramName);
+		if (sqlType != TYPE_UNKNOWN) {
+			return sqlType;
+		}
+		Object desc = getDescriptor(paramName);
+		if (desc instanceof PropertyDescriptor pd) {
+			return StatementCreatorUtils.javaTypeToSqlParameterType(pd.getPropertyType());
+		}
+		else if (desc instanceof Field field) {
+			return StatementCreatorUtils.javaTypeToSqlParameterType(field.getType());
+		}
+		return TYPE_UNKNOWN;
+	}
+
+	@Nullable
+	private Object getDescriptor(String paramName) {
+		return this.descriptorMap.computeIfAbsent(paramName, name -> {
+			Object pd = BeanUtils.getPropertyDescriptor(this.paramObject.getClass(), name);
+			if (pd == null) {
+				pd = ReflectionUtils.findField(this.paramObject.getClass(), name);
+			}
+			return pd;
+		});
+	}
+
+}

spring-jdbc/src/main/java/org/springframework/jdbc/core/simple/DefaultJdbcClient.java

@@ -0,0 +1,313 @@
+/*
+ * Copyright 2002-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.jdbc.core.simple;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Stream;
+
+import javax.sql.DataSource;
+
+import org.springframework.jdbc.core.JdbcOperations;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.core.ResultSetExtractor;
+import org.springframework.jdbc.core.RowCallbackHandler;
+import org.springframework.jdbc.core.RowMapper;
+import org.springframework.jdbc.core.SqlParameterValue;
+import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
+import org.springframework.jdbc.core.namedparam.NamedParameterJdbcOperations;
+import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
+import org.springframework.jdbc.core.namedparam.SimplePropertySqlParameterSource;
+import org.springframework.jdbc.core.namedparam.SqlParameterSource;
+import org.springframework.jdbc.support.KeyHolder;
+import org.springframework.jdbc.support.rowset.SqlRowSet;
+import org.springframework.util.Assert;
+
+/**
+ * The default implementation of {@link JdbcClient},
+ * as created by the static factory methods.
+ *
+ * @author Juergen Hoeller
+ * @since 6.1
+ * @see JdbcClient#create(DataSource)
+ * @see JdbcClient#create(JdbcOperations)
+ * @see JdbcClient#create(NamedParameterJdbcOperations)
+ */
+class DefaultJdbcClient implements JdbcClient {
+
+	private final JdbcOperations classicOps;
+
+	private final NamedParameterJdbcOperations namedParamOps;
+
+
+	public DefaultJdbcClient(DataSource dataSource) {
+		this.classicOps = new JdbcTemplate(dataSource);
+		this.namedParamOps = new NamedParameterJdbcTemplate(this.classicOps);
+	}
+
+	public DefaultJdbcClient(JdbcOperations jdbcTemplate) {
+		Assert.notNull(jdbcTemplate, "JdbcTemplate must not be null");
+		this.classicOps = jdbcTemplate;
+		this.namedParamOps = new NamedParameterJdbcTemplate(jdbcTemplate);
+	}
+
+	public DefaultJdbcClient(NamedParameterJdbcOperations jdbcTemplate) {
+		Assert.notNull(jdbcTemplate, "JdbcTemplate must not be null");
+		this.classicOps = jdbcTemplate.getJdbcOperations();
+		this.namedParamOps = jdbcTemplate;
+	}
+
+
+	@Override
+	public StatementSpec sql(String sql) {
+		return new DefaultStatementSpec(sql);
+	}
+
+
+	private class DefaultStatementSpec implements StatementSpec {
+
+		private final String sql;
+
+		private final List<Object> indexedParams = new ArrayList<>();
+
+		private final MapSqlParameterSource namedParams = new MapSqlParameterSource();
+
+		private SqlParameterSource namedParamSource = this.namedParams;
+
+		public DefaultStatementSpec(String sql) {
+			this.sql = sql;
+		}
+
+		@Override
+		public StatementSpec param(Object value) {
+			this.indexedParams.add(value);
+			return this;
+		}
+
+		@Override
+		public StatementSpec param(int jdbcIndex, Object value) {
+			if (jdbcIndex < 1) {
+				throw new IllegalArgumentException("Invalid JDBC index: needs to start at 1");
+			}
+			int index = jdbcIndex - 1;
+			int size = this.indexedParams.size();
+			if (index < size) {
+				this.indexedParams.set(index, value);
+			}
+			else {
+				for (int i = size; i < index; i++) {
+					this.indexedParams.add(null);
+				}
+				this.indexedParams.add(value);
+			}
+			return this;
+		}
+
+		@Override
+		public StatementSpec param(int jdbcIndex, Object value, int sqlType) {
+			return param(jdbcIndex, new SqlParameterValue(sqlType, value));
+		}
+
+		@Override
+		public StatementSpec param(String name, Object value) {
+			this.namedParams.addValue(name, value);
+			return this;
+		}
+
+		@Override
+		public StatementSpec param(String name, Object value, int sqlType) {
+			this.namedParams.addValue(name, value, sqlType);
+			return this;
+		}
+
+		@Override
+		public StatementSpec params(List<?> values) {
+			this.indexedParams.addAll(values);
+			return this;
+		}
+
+		@Override
+		public StatementSpec params(Map<String, ?> paramMap) {
+			this.namedParams.addValues(paramMap);
+			return this;
+		}
+
+		@SuppressWarnings({"rawtypes", "unchecked"})
+		@Override
+		public StatementSpec paramSource(Object namedParamObject) {
+			this.namedParamSource = (namedParamObject instanceof Map map ?
+					new MapSqlParameterSource(map) :
+					new SimplePropertySqlParameterSource(namedParamObject));
+			return this;
+		}
+
+		@Override
+		public StatementSpec paramSource(SqlParameterSource namedParamSource) {
+			this.namedParamSource = namedParamSource;
+			return this;
+		}
+
+		@Override
+		public ResultQuerySpec query() {
+			return (useNamedParams() ?
+					new NamedParamResultQuerySpec() :
+					new IndexedParamResultQuerySpec());
+		}
+
+		@Override
+		public <T> MappedQuerySpec<T> query(RowMapper<T> rowMapper) {
+			return (useNamedParams() ?
+					new NamedParamMappedQuerySpec<>(rowMapper) :
+					new IndexedParamMappedQuerySpec<>(rowMapper));
+		}
+
+		@Override
+		public void query(RowCallbackHandler rch) {
+			if (useNamedParams()) {
+				namedParamOps.query(this.sql, this.namedParams, rch);
+			}
+			else {
+				classicOps.query(this.sql, rch, this.indexedParams.toArray());
+			}
+		}
+
+		@Override
+		public <T> T query(ResultSetExtractor<T> rse) {
+			T result = (useNamedParams() ?
+					namedParamOps.query(this.sql, this.namedParams, rse) :
+					classicOps.query(this.sql, rse, this.indexedParams.toArray()));
+			Assert.state(result != null, "No result from ResultSetExtractor");
+			return result;
+		}
+
+		@Override
+		public int update() {
+			return (useNamedParams() ?
+					namedParamOps.update(this.sql, this.namedParamSource) :
+					classicOps.update(this.sql, this.indexedParams.toArray()));
+		}
+
+		@Override
+		public int update(KeyHolder generatedKeyHolder) {
+			return (useNamedParams() ?
+					namedParamOps.update(this.sql, this.namedParamSource, generatedKeyHolder) :
+					classicOps.update(this.sql, this.indexedParams.toArray(), generatedKeyHolder));
+		}
+
+		private boolean useNamedParams() {
+			boolean hasNamedParams = (this.namedParams.hasValues() || this.namedParamSource != this.namedParams);
+			if (hasNamedParams && !this.indexedParams.isEmpty()) {
+				throw new IllegalStateException("Configure either named or indexed parameters, not both");
+			}
+			if (this.namedParams.hasValues() && this.namedParamSource != this.namedParams) {
+				throw new IllegalStateException(
+						"Configure either individual named parameters or a SqlParameterSource, not both");
+			}
+			return hasNamedParams;
+		}
+
+
+		private class IndexedParamResultQuerySpec implements ResultQuerySpec {
+
+			@Override
+			public SqlRowSet rowSet() {
+				return classicOps.queryForRowSet(sql, indexedParams.toArray());
+			}
+
+			@Override
+			public List<Map<String, Object>> listOfRows() {
+				return classicOps.queryForList(sql, indexedParams.toArray());
+			}
+
+			@Override
+			public Map<String, Object> singleRow() {
+				return classicOps.queryForMap(sql, indexedParams.toArray());
+			}
+
+			@Override
+			public <T> List<T> singleColumn(Class<T> requiredType) {
+				return classicOps.queryForList(sql, requiredType, indexedParams.toArray());
+			}
+		}
+
+
+		private class NamedParamResultQuerySpec implements ResultQuerySpec {
+
+			@Override
+			public SqlRowSet rowSet() {
+				return namedParamOps.queryForRowSet(sql, namedParamSource);
+			}
+
+			@Override
+			public List<Map<String, Object>> listOfRows() {
+				return namedParamOps.queryForList(sql, namedParamSource);
+			}
+
+			@Override
+			public Map<String, Object> singleRow() {
+				return namedParamOps.queryForMap(sql, namedParamSource);
+			}
+
+			@Override
+			public <T> List<T> singleColumn(Class<T> requiredType) {
+				return namedParamOps.queryForList(sql, namedParamSource, requiredType);
+			}
+		}
+
+
+		private class IndexedParamMappedQuerySpec<T> implements MappedQuerySpec<T> {
+
+			private final RowMapper<T> rowMapper;
+
+			public IndexedParamMappedQuerySpec(RowMapper<T> rowMapper) {
+				this.rowMapper = rowMapper;
+			}
+
+			@Override
+			public Stream<T> stream() {
+				return classicOps.queryForStream(sql, this.rowMapper, indexedParams.toArray());
+			}
+
+			@Override
+			public List<T> list() {
+				return classicOps.query(sql, this.rowMapper, indexedParams.toArray());
+			}
+		}
+
+
+		private class NamedParamMappedQuerySpec<T> implements MappedQuerySpec<T> {
+
+			private final RowMapper<T> rowMapper;
+
+			public NamedParamMappedQuerySpec(RowMapper<T> rowMapper) {
+				this.rowMapper = rowMapper;
+			}
+
+			@Override
+			public Stream<T> stream() {
+				return namedParamOps.queryForStream(sql, namedParamSource, this.rowMapper);
+			}
+
+			@Override
+			public List<T> list() {
+				return namedParamOps.query(sql, namedParamSource, this.rowMapper);
+			}
+		}
+	}
+
+}

spring-jdbc/src/main/java/org/springframework/jdbc/core/simple/JdbcClient.java

@@ -0,0 +1,366 @@
+/*
+ * Copyright 2002-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.jdbc.core.simple;
+
+import java.util.Collection;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Stream;
+
+import javax.sql.DataSource;
+
+import org.springframework.dao.support.DataAccessUtils;
+import org.springframework.jdbc.core.JdbcOperations;
+import org.springframework.jdbc.core.ResultSetExtractor;
+import org.springframework.jdbc.core.RowCallbackHandler;
+import org.springframework.jdbc.core.RowMapper;
+import org.springframework.jdbc.core.namedparam.NamedParameterJdbcOperations;
+import org.springframework.jdbc.core.namedparam.SqlParameterSource;
+import org.springframework.jdbc.support.KeyHolder;
+import org.springframework.jdbc.support.rowset.SqlRowSet;
+
+/**
+ * A fluent {@link JdbcClient} with common JDBC query and update operations,
+ * supporting JDBC-style positional as well as Spring-style named parameters
+ * with a convenient unified facade for JDBC PreparedStatement execution.
+ *
+ * <p>An example for retrieving a query result as a {@code java.util.Optional}:
+ * <pre class="code">
+ * Optional&lt;Integer&gt; value = client.sql("SELECT AGE FROM CUSTMR WHERE ID = :id")
+ *     .param("id", 3)
+ *     .query((rs, rowNum) -> rs.getInt(1))
+ *     .optional();
+ * </pre>
+ *
+ * <p>Delegates to {@link org.springframework.jdbc.core.JdbcTemplate} and
+ * {@link org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate}.
+ * For complex JDBC operations, e.g. batch inserts and stored procedure calls,
+ * you may use those lower-level template classes directly - or alternatively,
+ * {@link SimpleJdbcInsert} and {@link SimpleJdbcCall}.
+ *
+ * @author Juergen Hoeller
+ * @since 6.1
+ * @see ResultSetExtractor
+ * @see RowCallbackHandler
+ * @see RowMapper
+ * @see JdbcOperations
+ * @see NamedParameterJdbcOperations
+ * @see org.springframework.jdbc.core.JdbcTemplate
+ * @see org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate
+ */
+public interface JdbcClient {
+
+	/**
+	 * The starting point for any JDBC operation: a custom SQL String.
+	 * @param sql the SQL query or update statement as a String
+	 * @return a chained statement specification
+	 */
+	StatementSpec sql(String sql);
+
+
+	// Static factory methods
+
+	/**
+	 * Create a {@code JdbcClient} for the given {@link DataSource}.
+	 * @param dataSource the DataSource to obtain connections from
+	 */
+	static JdbcClient create(DataSource dataSource) {
+		return new DefaultJdbcClient(dataSource);
+	}
+
+	/**
+	 * Create a {@code JdbcClient} for the given {@link JdbcOperations} delegate,
+	 * typically a {@link org.springframework.jdbc.core.JdbcTemplate}.
+	 * <p>Use this factory method for reusing existing {@code JdbcTemplate} configuration,
+	 * including its {@code DataSource}.
+	 * @param jdbcTemplate the delegate to perform operations on
+	 */
+	static JdbcClient create(JdbcOperations jdbcTemplate) {
+		return new DefaultJdbcClient(jdbcTemplate);
+	}
+
+	/**
+	 * Create a {@code JdbcClient} for the given {@link NamedParameterJdbcOperations} delegate,
+	 * typically a {@link org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate}.
+	 * <p>Use this factory method for reusing existing {@code NamedParameterJdbcTemplate}
+	 * configuration, including its underlying {@code JdbcTemplate} and the {@code DataSource}.
+	 * @param jdbcTemplate the delegate to perform operations on
+	 */
+	static JdbcClient create(NamedParameterJdbcOperations jdbcTemplate) {
+		return new DefaultJdbcClient(jdbcTemplate);
+	}
+
+
+	/**
+	 * A statement specification for parameter bindings and query/update execution.
+	 */
+	interface StatementSpec {
+
+		/**
+		 * Bind a positional JDBC statement parameter for "?" placeholder resolution
+		 * by implicit order of parameter value registration.
+		 * <p>This is primarily intended for statements with a single or very few
+		 * parameters, registering each parameter value in the order of the
+		 * parameter's occurrence in the SQL statement.
+		 * @param value the parameter value to bind
+		 * @return this statement specification (for chaining)
+		 * @see java.sql.PreparedStatement#setObject(int, Object)
+		 */
+		StatementSpec param(Object value);
+
+		/**
+		 * Bind a positional JDBC statement parameter for "?" placeholder resolution
+		 * by explicit JDBC statement parameter index.
+		 * @param jdbcIndex the JDBC-style index (starting with 1)
+		 * @param value the parameter value to bind
+		 * @return this statement specification (for chaining)
+		 * @see java.sql.PreparedStatement#setObject(int, Object)
+		 */
+		StatementSpec param(int jdbcIndex, Object value);
+
+		/**
+		 * Bind a positional JDBC statement parameter for "?" placeholder resolution
+		 * by explicit JDBC statement parameter index.
+		 * @param jdbcIndex the JDBC-style index (starting with 1)
+		 * @param value the parameter value to bind
+		 * @param sqlType the associated SQL type (see {@link java.sql.Types})
+		 * @return this statement specification (for chaining)
+		 * @see java.sql.PreparedStatement#setObject(int, Object, int)
+		 */
+		StatementSpec param(int jdbcIndex, Object value, int sqlType);
+
+		/**
+		 * Bind a named statement parameter for ":x" placeholder resolution,
+		 * with each "x" name matching a ":x" placeholder in the SQL statement.
+		 * @param name the parameter name
+		 * @param value the parameter value to bind
+		 * @return this statement specification (for chaining)
+		 * @see org.springframework.jdbc.core.namedparam.MapSqlParameterSource#addValue(String, Object)
+		 */
+		StatementSpec param(String name, Object value);
+
+		/**
+		 * Bind a named statement parameter for ":x" placeholder resolution,
+		 * with each "x" name matching a ":x" placeholder in the SQL statement.
+		 * @param name the parameter name
+		 * @param value the parameter value to bind
+		 * @param sqlType the associated SQL type (see {@link java.sql.Types})
+		 * @return this statement specification (for chaining)
+		 * @see org.springframework.jdbc.core.namedparam.MapSqlParameterSource#addValue(String, Object, int)
+		 */
+		StatementSpec param(String name, Object value, int sqlType);
+
+		/**
+		 * Bind a list of positional parameters for "?" placeholder resolution.
+		 * <p>The given list will be added to existing positional parameters, if any.
+		 * Each element from the complete list will be bound as a JDBC positional
+		 * parameter with a corresponding JDBC index (i.e. list index + 1).
+		 * @param values the parameter values to bind
+		 * @return this statement specification (for chaining)
+		 * @see #param(Object)
+		 */
+		StatementSpec params(List<?> values);
+
+		/**
+		 * Bind named statement parameters for ":x" placeholder resolution.
+		 * <p>The given map will be merged into existing named parameters, if any.
+		 * @param paramMap a map of names and parameter values to bind
+		 * @return this statement specification (for chaining)
+		 * @see #param(String, Object)
+		 */
+		StatementSpec params(Map<String, ?> paramMap);
+
+		/**
+		 * Bind named statement parameters for ":x" placeholder resolution.
+		 * <p>The given parameter object will define all named parameters
+		 * based on its JavaBean properties, record components or raw fields.
+		 * A Map instance can be provided as a complete parameter source as well.
+		 * @param namedParamObject a custom parameter object (e.g. a JavaBean or
+		 * record class) with named properties serving as statement parameters
+		 * @return this statement specification (for chaining)
+		 * @see org.springframework.jdbc.core.namedparam.MapSqlParameterSource
+		 * @see org.springframework.jdbc.core.namedparam.SimplePropertySqlParameterSource
+		 */
+		StatementSpec paramSource(Object namedParamObject);
+
+		/**
+		 * Bind named statement parameters for ":x" placeholder resolution.
+		 * <p>The given parameter source will define all named parameters,
+		 * possibly associating specific SQL types with each value.
+		 * @param namedParamSource a custom {@link SqlParameterSource} instance
+		 * @return this statement specification (for chaining)
+		 * @see org.springframework.jdbc.core.namedparam.AbstractSqlParameterSource#registerSqlType
+		 */
+		StatementSpec paramSource(SqlParameterSource namedParamSource);
+
+		/**
+		 * Proceed towards execution of a query, with several result options
+		 * available in the returned query specification.
+		 * @return the result query specification
+		 * @see java.sql.PreparedStatement#executeQuery()
+		 */
+		ResultQuerySpec query();
+
+		/**
+		 * Proceed towards execution of a mapped query, with several options
+		 * available in the returned query specification.
+		 * @param rowMapper the callback for mapping each row in the ResultSet
+		 * @return the mapped query specification
+		 * @see java.sql.PreparedStatement#executeQuery()
+		 */
+		<T> MappedQuerySpec<T> query(RowMapper<T> rowMapper);
+
+		/**
+		 * Execute a query with the provided SQL statement,
+		 * processing each row with the given callback.
+		 * @param rch a callback for processing each row in the ResultSet
+		 * @see java.sql.PreparedStatement#executeQuery()
+		 */
+		void query(RowCallbackHandler rch);
+
+		/**
+		 * Execute a query with the provided SQL statement,
+		 * returning a result object for the entire ResultSet.
+		 * @param rse a callback for processing the entire ResultSet
+		 * @return the value returned by the ResultSetExtractor
+		 * @see java.sql.PreparedStatement#executeQuery()
+		 */
+		<T> T query(ResultSetExtractor<T> rse);
+
+		/**
+		 * Execute the provided SQL statement as an update.
+		 * @return the number of rows affected
+		 * @see java.sql.PreparedStatement#executeUpdate()
+		 */
+		int update();
+
+		/**
+		 * Execute the provided SQL statement as an update.
+		 * @param generatedKeyHolder a KeyHolder that will hold the generated keys
+		 * (typically a {@link org.springframework.jdbc.support.GeneratedKeyHolder})
+		 * @return the number of rows affected
+		 * @see java.sql.PreparedStatement#executeUpdate()
+		 */
+		int update(KeyHolder generatedKeyHolder);
+	}
+
+
+	/**
+	 * A specification for simple result queries.
+	 */
+	interface ResultQuerySpec {
+
+		/**
+		 * Retrieve the result as a row set.
+		 * @return a detached row set representation
+		 * of the original database result
+		 */
+		SqlRowSet rowSet();
+
+		/**
+		 * Retrieve the result as a list of rows,
+		 * retaining the order from the original database result.
+		 * @return a (potentially empty) list of rows,
+		 * with each result row represented as a map of
+		 * case-insensitive column names to column values
+		 */
+		List<Map<String, Object>> listOfRows();
+
+		/**
+		 * Retrieve a single row result.
+		 * @return the result row represented as a map of
+		 * case-insensitive column names to column values
+		 */
+		Map<String, Object> singleRow();
+
+		/**
+		 * Retrieve a single column result,
+		 * retaining the order from the original database result.
+		 * @return a (potentially empty) list of rows, with each
+		 * row represented as a column value of the given type
+		 */
+		<T> List<T> singleColumn(Class<T> requiredType);
+
+		/**
+		 * Retrieve a single value result.
+		 * @return the single row represented as its single
+		 * column value of the given type
+		 * @see DataAccessUtils#requiredSingleResult(Collection)
+		 */
+		default <T> T singleValue(Class<T> requiredType) {
+			return DataAccessUtils.requiredSingleResult(singleColumn(requiredType));
+		}
+	}
+
+
+	/**
+	 * A specification for RowMapper-mapped queries.
+	 *
+	 * @param <T> the RowMapper-declared result type
+	 */
+	interface MappedQuerySpec<T> {
+
+		/**
+		 * Retrieve the result as a lazily resolved stream of mapped objects,
+		 * retaining the order from the original database result.
+		 * @return the result Stream, containing mapped objects, needing to be
+		 * closed once fully processed (e.g. through a try-with-resources clause)
+		 */
+		Stream<T> stream();
+
+		/**
+		 * Retrieve the result as a pre-resolved list of mapped objects,
+		 * retaining the order from the original database result.
+		 * @return the result as a detached List, containing mapped objects
+		 */
+		List<T> list();
+
+		/**
+		 * Retrieve the result as an order-preserving set of mapped objects.
+		 * @return the result as a detached Set, containing mapped objects
+		 * @see #list()
+		 * @see LinkedHashSet
+		 */
+		default Set<T> set() {
+			return new LinkedHashSet<>(list());
+		}
+
+		/**
+		 * Retrieve a single result, if available, as an {@link Optional} handle.
+		 * @return an Optional handle with a single result object or none
+		 * @see #list()
+		 * @see DataAccessUtils#optionalResult(Collection)
+		 */
+		default Optional<T> optional() {
+			return DataAccessUtils.optionalResult(list());
+		}
+
+		/**
+		 * Retrieve a single result as a required object instance.
+		 * @return the single result object (never {@code null})
+		 * @see #list()
+		 * @see DataAccessUtils#requiredSingleResult(Collection)
+		 */
+		default T single() {
+			return DataAccessUtils.requiredSingleResult(list());
+		}
+	}
+
+}

spring-jdbc/src/test/java/org/springframework/jdbc/core/namedparam/NamedParameterQueryTests.java

@@ -34,8 +34,6 @@ import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
-import org.springframework.jdbc.core.RowMapper;
-
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.BDDMockito.given;
@@ -129,8 +127,7 @@ public class NamedParameterQueryTests {
 	}
 
 	@Test
-	public void testQueryForListWithParamMapAndIntegerElementAndSingleRowAndColumn()
-			throws Exception {
+	public void testQueryForListWithParamMapAndIntegerElementAndSingleRowAndColumn() throws Exception {
 		given(resultSet.getMetaData()).willReturn(resultSetMetaData);
 		given(resultSet.next()).willReturn(true, false);
 		given(resultSet.getInt(1)).willReturn(11);
@@ -168,11 +165,10 @@ public class NamedParameterQueryTests {
 
 		MapSqlParameterSource params = new MapSqlParameterSource();
 		params.addValue("id", 3);
-		Object o = template.queryForObject("SELECT AGE FROM CUSTMR WHERE ID = :id",
-				params, (RowMapper<Object>) (rs, rowNum) -> rs.getInt(1));
+		Integer value = template.queryForObject("SELECT AGE FROM CUSTMR WHERE ID = :id",
+				params, (rs, rowNum) -> rs.getInt(1));
 
-		boolean condition = o instanceof Integer;
-		assertThat(condition).as("Correct result type").isTrue();
+		assertThat(value).isEqualTo(22);
 		verify(connection).prepareStatement("SELECT AGE FROM CUSTMR WHERE ID = ?");
 		verify(preparedStatement).setObject(1, 3);
 	}
@@ -185,11 +181,10 @@ public class NamedParameterQueryTests {
 
 		Map<String, Object> params = new HashMap<>();
 		params.put("id", 3);
-		Object o = template.queryForObject("SELECT AGE FROM CUSTMR WHERE ID = :id",
+		Integer value = template.queryForObject("SELECT AGE FROM CUSTMR WHERE ID = :id",
 				params, Integer.class);
 
-		boolean condition = o instanceof Integer;
-		assertThat(condition).as("Correct result type").isTrue();
+		assertThat(value).isEqualTo(22);
 		verify(connection).prepareStatement("SELECT AGE FROM CUSTMR WHERE ID = ?");
 		verify(preparedStatement).setObject(1, 3);
 	}
@@ -202,30 +197,26 @@ public class NamedParameterQueryTests {
 
 		MapSqlParameterSource params = new MapSqlParameterSource();
 		params.addValue("id", 3);
-		Object o = template.queryForObject("SELECT AGE FROM CUSTMR WHERE ID = :id",
+		Integer value = template.queryForObject("SELECT AGE FROM CUSTMR WHERE ID = :id",
 				params, Integer.class);
 
-		boolean condition = o instanceof Integer;
-		assertThat(condition).as("Correct result type").isTrue();
+		assertThat(value).isEqualTo(22);
 		verify(connection).prepareStatement("SELECT AGE FROM CUSTMR WHERE ID = ?");
 		verify(preparedStatement).setObject(1, 3);
 	}
 
 	@Test
 	public void testQueryForObjectWithParamMapAndList() throws Exception {
-		String sql = "SELECT AGE FROM CUSTMR WHERE ID IN (:ids)";
-		String sqlToUse = "SELECT AGE FROM CUSTMR WHERE ID IN (?, ?)";
 		given(resultSet.getMetaData()).willReturn(resultSetMetaData);
 		given(resultSet.next()).willReturn(true, false);
 		given(resultSet.getInt(1)).willReturn(22);
 
 		MapSqlParameterSource params = new MapSqlParameterSource();
 		params.addValue("ids", Arrays.asList(3, 4));
-		Object o = template.queryForObject(sql, params, Integer.class);
+		Integer value = template.queryForObject("SELECT AGE FROM CUSTMR WHERE ID IN (:ids)", params, Integer.class);
 
-		boolean condition = o instanceof Integer;
-		assertThat(condition).as("Correct result type").isTrue();
-		verify(connection).prepareStatement(sqlToUse);
+		assertThat(value).isEqualTo(22);
+		verify(connection).prepareStatement("SELECT AGE FROM CUSTMR WHERE ID IN (?, ?)");
 		verify(preparedStatement).setObject(1, 3);
 	}
 
@@ -240,14 +231,11 @@ public class NamedParameterQueryTests {
 		l1.add(new Object[] {3, "Rod"});
 		l1.add(new Object[] {4, "Juergen"});
 		params.addValue("multiExpressionList", l1);
-		Object o = template.queryForObject(
-				"SELECT AGE FROM CUSTMR WHERE (ID, NAME) IN (:multiExpressionList)",
+		Integer value = template.queryForObject("SELECT AGE FROM CUSTMR WHERE (ID, NAME) IN (:multiExpressionList)",
 				params, Integer.class);
 
-		boolean condition = o instanceof Integer;
-		assertThat(condition).as("Correct result type").isTrue();
-		verify(connection).prepareStatement(
-				"SELECT AGE FROM CUSTMR WHERE (ID, NAME) IN ((?, ?), (?, ?))");
+		assertThat(value).isEqualTo(22);
+		verify(connection).prepareStatement("SELECT AGE FROM CUSTMR WHERE (ID, NAME) IN ((?, ?), (?, ?))");
 		verify(preparedStatement).setObject(1, 3);
 	}
 
@@ -295,6 +283,20 @@ public class NamedParameterQueryTests {
 		verify(preparedStatement).setObject(2, 5);
 	}
 
+	@Test
+	public void testQueryForLongWithParamRecord() throws Exception {
+		given(resultSet.getMetaData()).willReturn(resultSetMetaData);
+		given(resultSet.next()).willReturn(true, false);
+		given(resultSet.getLong(1)).willReturn(87L);
+
+		BeanPropertySqlParameterSource params = new BeanPropertySqlParameterSource(new ParameterRecord(3));
+		long l = template.queryForObject("SELECT AGE FROM CUSTMR WHERE ID = :id", params, Long.class);
+
+		assertThat(l).as("Return of a long").isEqualTo(87);
+		verify(connection).prepareStatement("SELECT AGE FROM CUSTMR WHERE ID = ?");
+		verify(preparedStatement).setObject(1, 3, Types.INTEGER);
+	}
+
 
 	static class ParameterBean {
 
@@ -323,4 +325,8 @@ public class NamedParameterQueryTests {
 		}
 	}
 
+
+	record ParameterRecord(int id) {
+	}
+
 }

spring-jdbc/src/test/java/org/springframework/jdbc/core/simple/JdbcClientIndexedParameterTests.java

@@ -0,0 +1,332 @@
+/*
+ * Copyright 2002-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.jdbc.core.simple;
+
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Types;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Stream;
+
+import javax.sql.DataSource;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.jdbc.Customer;
+import org.springframework.jdbc.core.SqlParameterValue;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+/**
+ * @author Juergen Hoeller
+ * @since 6.1
+ */
+public class JdbcClientIndexedParameterTests {
+
+	private static final String SELECT_NAMED_PARAMETERS =
+			"select id, forename from custmr where id = ? and country = ?";
+	private static final String SELECT_NO_PARAMETERS =
+			"select id, forename from custmr";
+
+	private static final String UPDATE_NAMED_PARAMETERS =
+			"update seat_status set booking_id = null where performance_id = ? and price_band_id = ?";
+
+	private static final String[] COLUMN_NAMES = new String[] {"id", "forename"};
+
+
+	private Connection connection = mock();
+
+	private DataSource dataSource = mock();
+
+	private PreparedStatement preparedStatement = mock();
+
+	private ResultSet resultSet = mock();
+
+	private DatabaseMetaData databaseMetaData = mock();
+
+	private JdbcClient client = JdbcClient.create(dataSource);
+
+	private List<Object> params = new ArrayList<>();
+
+
+	@BeforeEach
+	public void setup() throws Exception {
+		given(dataSource.getConnection()).willReturn(connection);
+		given(connection.prepareStatement(anyString())).willReturn(preparedStatement);
+		given(preparedStatement.getConnection()).willReturn(connection);
+		given(preparedStatement.executeQuery()).willReturn(resultSet);
+		given(databaseMetaData.getDatabaseProductName()).willReturn("MySQL");
+		given(databaseMetaData.supportsBatchUpdates()).willReturn(true);
+	}
+
+
+	@Test
+	public void testQueryWithResultSetExtractor() throws SQLException {
+		given(resultSet.next()).willReturn(true);
+		given(resultSet.getInt("id")).willReturn(1);
+		given(resultSet.getString("forename")).willReturn("rod");
+
+		params.add(new SqlParameterValue(Types.DECIMAL, 1));
+		params.add("UK");
+		Customer cust = client.sql(SELECT_NAMED_PARAMETERS).params(params).query(
+				rs -> {
+					rs.next();
+					Customer cust1 = new Customer();
+					cust1.setId(rs.getInt(COLUMN_NAMES[0]));
+					cust1.setForename(rs.getString(COLUMN_NAMES[1]));
+					return cust1;
+				});
+
+		assertThat(cust.getId()).as("Customer id was assigned correctly").isEqualTo(1);
+		assertThat(cust.getForename()).as("Customer forename was assigned correctly").isEqualTo("rod");
+		verify(connection).prepareStatement(SELECT_NAMED_PARAMETERS);
+		verify(preparedStatement).setObject(1, 1, Types.DECIMAL);
+		verify(preparedStatement).setString(2, "UK");
+		verify(resultSet).close();
+		verify(preparedStatement).close();
+		verify(connection).close();
+	}
+
+	@Test
+	public void testQueryWithResultSetExtractorNoParameters() throws SQLException {
+		given(resultSet.next()).willReturn(true);
+		given(resultSet.getInt("id")).willReturn(1);
+		given(resultSet.getString("forename")).willReturn("rod");
+
+		Customer cust = client.sql(SELECT_NO_PARAMETERS).query(
+				rs -> {
+					rs.next();
+					Customer cust1 = new Customer();
+					cust1.setId(rs.getInt(COLUMN_NAMES[0]));
+					cust1.setForename(rs.getString(COLUMN_NAMES[1]));
+					return cust1;
+				});
+
+		assertThat(cust.getId()).as("Customer id was assigned correctly").isEqualTo(1);
+		assertThat(cust.getForename()).as("Customer forename was assigned correctly").isEqualTo("rod");
+		verify(connection).prepareStatement(SELECT_NO_PARAMETERS);
+		verify(resultSet).close();
+		verify(preparedStatement).close();
+		verify(connection).close();
+	}
+
+	@Test
+	public void testQueryWithRowCallbackHandler() throws SQLException {
+		given(resultSet.next()).willReturn(true, false);
+		given(resultSet.getInt("id")).willReturn(1);
+		given(resultSet.getString("forename")).willReturn("rod");
+
+		params.add(new SqlParameterValue(Types.DECIMAL, 1));
+		params.add("UK");
+		final List<Customer> customers = new ArrayList<>();
+		client.sql(SELECT_NAMED_PARAMETERS).params(params).query(rs -> {
+			Customer cust = new Customer();
+			cust.setId(rs.getInt(COLUMN_NAMES[0]));
+			cust.setForename(rs.getString(COLUMN_NAMES[1]));
+			customers.add(cust);
+		});
+
+		assertThat(customers).hasSize(1);
+		assertThat(customers.get(0).getId()).as("Customer id was assigned correctly").isEqualTo(1);
+		assertThat(customers.get(0).getForename()).as("Customer forename was assigned correctly").isEqualTo("rod");
+		verify(connection).prepareStatement(SELECT_NAMED_PARAMETERS);
+		verify(preparedStatement).setObject(1, 1, Types.DECIMAL);
+		verify(preparedStatement).setString(2, "UK");
+		verify(resultSet).close();
+		verify(preparedStatement).close();
+		verify(connection).close();
+	}
+
+	@Test
+	public void testQueryWithRowCallbackHandlerNoParameters() throws SQLException {
+		given(resultSet.next()).willReturn(true, false);
+		given(resultSet.getInt("id")).willReturn(1);
+		given(resultSet.getString("forename")).willReturn("rod");
+
+		final List<Customer> customers = new ArrayList<>();
+		client.sql(SELECT_NO_PARAMETERS).query(rs -> {
+			Customer cust = new Customer();
+			cust.setId(rs.getInt(COLUMN_NAMES[0]));
+			cust.setForename(rs.getString(COLUMN_NAMES[1]));
+			customers.add(cust);
+		});
+
+		assertThat(customers).hasSize(1);
+		assertThat(customers.get(0).getId()).as("Customer id was assigned correctly").isEqualTo(1);
+		assertThat(customers.get(0).getForename()).as("Customer forename was assigned correctly").isEqualTo("rod");
+		verify(connection).prepareStatement(SELECT_NO_PARAMETERS);
+		verify(resultSet).close();
+		verify(preparedStatement).close();
+		verify(connection).close();
+	}
+
+	@Test
+	public void testQueryWithRowMapper() throws SQLException {
+		given(resultSet.next()).willReturn(true, false);
+		given(resultSet.getInt("id")).willReturn(1);
+		given(resultSet.getString("forename")).willReturn("rod");
+
+		params.add(new SqlParameterValue(Types.DECIMAL, 1));
+		params.add("UK");
+		List<Customer> customers = client.sql(SELECT_NAMED_PARAMETERS).params(params).query(
+				(rs, rownum) -> {
+					Customer cust = new Customer();
+					cust.setId(rs.getInt(COLUMN_NAMES[0]));
+					cust.setForename(rs.getString(COLUMN_NAMES[1]));
+					return cust;
+				}).list();
+
+		assertThat(customers).hasSize(1);
+		Customer cust = customers.get(0);
+		assertThat(cust.getId()).as("Customer id was assigned correctly").isEqualTo(1);
+		assertThat(cust.getForename()).as("Customer forename was assigned correctly").isEqualTo("rod");
+		verify(connection).prepareStatement(SELECT_NAMED_PARAMETERS);
+		verify(preparedStatement).setObject(1, 1, Types.DECIMAL);
+		verify(preparedStatement).setString(2, "UK");
+		verify(resultSet).close();
+		verify(preparedStatement).close();
+		verify(connection).close();
+	}
+
+	@Test
+	public void testQueryWithRowMapperNoParameters() throws SQLException {
+		given(resultSet.next()).willReturn(true, false);
+		given(resultSet.getInt("id")).willReturn(1);
+		given(resultSet.getString("forename")).willReturn("rod");
+
+		Set<Customer> customers = client.sql(SELECT_NO_PARAMETERS).query(
+				(rs, rownum) -> {
+					Customer cust = new Customer();
+					cust.setId(rs.getInt(COLUMN_NAMES[0]));
+					cust.setForename(rs.getString(COLUMN_NAMES[1]));
+					return cust;
+				}).set();
+
+		assertThat(customers).hasSize(1);
+		Customer cust = customers.iterator().next();
+		assertThat(cust.getId()).as("Customer id was assigned correctly").isEqualTo(1);
+		assertThat(cust.getForename()).as("Customer forename was assigned correctly").isEqualTo("rod");
+		verify(connection).prepareStatement(SELECT_NO_PARAMETERS);
+		verify(resultSet).close();
+		verify(preparedStatement).close();
+		verify(connection).close();
+	}
+
+	@Test
+	public void testQueryForObjectWithRowMapper() throws SQLException {
+		given(resultSet.next()).willReturn(true, false);
+		given(resultSet.getInt("id")).willReturn(1);
+		given(resultSet.getString("forename")).willReturn("rod");
+
+		params.add(new SqlParameterValue(Types.DECIMAL, 1));
+		params.add("UK");
+
+		Customer cust = client.sql(SELECT_NAMED_PARAMETERS).params(params).query(
+				(rs, rownum) -> {
+					Customer cust1 = new Customer();
+					cust1.setId(rs.getInt(COLUMN_NAMES[0]));
+					cust1.setForename(rs.getString(COLUMN_NAMES[1]));
+					return cust1;
+				}).single();
+
+		assertThat(cust.getId()).as("Customer id was assigned correctly").isEqualTo(1);
+		assertThat(cust.getForename()).as("Customer forename was assigned correctly").isEqualTo("rod");
+		verify(connection).prepareStatement(SELECT_NAMED_PARAMETERS);
+		verify(preparedStatement).setObject(1, 1, Types.DECIMAL);
+		verify(preparedStatement).setString(2, "UK");
+		verify(resultSet).close();
+		verify(preparedStatement).close();
+		verify(connection).close();
+	}
+
+	@Test
+	public void testQueryForStreamWithRowMapper() throws SQLException {
+		given(resultSet.next()).willReturn(true, false);
+		given(resultSet.getInt("id")).willReturn(1);
+		given(resultSet.getString("forename")).willReturn("rod");
+
+		params.add(new SqlParameterValue(Types.DECIMAL, 1));
+		params.add("UK");
+		AtomicInteger count = new AtomicInteger();
+
+		try (Stream<Customer> s = client.sql(SELECT_NAMED_PARAMETERS).params(params).query(
+				(rs, rownum) -> {
+					Customer cust1 = new Customer();
+					cust1.setId(rs.getInt(COLUMN_NAMES[0]));
+					cust1.setForename(rs.getString(COLUMN_NAMES[1]));
+					return cust1;
+				}).stream()) {
+			s.forEach(cust -> {
+				count.incrementAndGet();
+				assertThat(cust.getId()).as("Customer id was assigned correctly").isEqualTo(1);
+				assertThat(cust.getForename()).as("Customer forename was assigned correctly").isEqualTo("rod");
+			});
+		}
+
+		assertThat(count.get()).isEqualTo(1);
+		verify(connection).prepareStatement(SELECT_NAMED_PARAMETERS);
+		verify(preparedStatement).setObject(1, 1, Types.DECIMAL);
+		verify(preparedStatement).setString(2, "UK");
+		verify(resultSet).close();
+		verify(preparedStatement).close();
+		verify(connection).close();
+	}
+
+	@Test
+	public void testUpdate() throws SQLException {
+		given(preparedStatement.executeUpdate()).willReturn(1);
+
+		params.add(1);
+		params.add(1);
+		int rowsAffected = client.sql(UPDATE_NAMED_PARAMETERS).params(params).update();
+
+		assertThat(rowsAffected).isEqualTo(1);
+		verify(connection).prepareStatement(UPDATE_NAMED_PARAMETERS);
+		verify(preparedStatement).setObject(1, 1);
+		verify(preparedStatement).setObject(2, 1);
+		verify(preparedStatement).close();
+		verify(connection).close();
+	}
+
+	@Test
+	public void testUpdateWithTypedParameters() throws SQLException {
+		given(preparedStatement.executeUpdate()).willReturn(1);
+
+		params.add(new SqlParameterValue(Types.DECIMAL, 1));
+		params.add(new SqlParameterValue(Types.INTEGER, 1));
+		int rowsAffected = client.sql(UPDATE_NAMED_PARAMETERS).params(params).update();
+
+		assertThat(rowsAffected).isEqualTo(1);
+		verify(connection).prepareStatement(UPDATE_NAMED_PARAMETERS);
+		verify(preparedStatement).setObject(1, 1, Types.DECIMAL);
+		verify(preparedStatement).setObject(2, 1, Types.INTEGER);
+		verify(preparedStatement).close();
+		verify(connection).close();
+	}
+
+}

spring-jdbc/src/test/java/org/springframework/jdbc/core/simple/JdbcClientNamedParameterTests.java

@@ -0,0 +1,338 @@
+/*
+ * Copyright 2002-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.jdbc.core.simple;
+
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Types;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Stream;
+
+import javax.sql.DataSource;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.jdbc.Customer;
+import org.springframework.jdbc.core.SqlParameterValue;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+/**
+ * @author Juergen Hoeller
+ * @since 6.1
+ */
+public class JdbcClientNamedParameterTests {
+
+	private static final String SELECT_NAMED_PARAMETERS =
+			"select id, forename from custmr where id = :id and country = :country";
+	private static final String SELECT_NAMED_PARAMETERS_PARSED =
+			"select id, forename from custmr where id = ? and country = ?";
+	private static final String SELECT_NO_PARAMETERS =
+			"select id, forename from custmr";
+
+	private static final String UPDATE_NAMED_PARAMETERS =
+			"update seat_status set booking_id = null where performance_id = :perfId and price_band_id = :priceId";
+	private static final String UPDATE_NAMED_PARAMETERS_PARSED =
+			"update seat_status set booking_id = null where performance_id = ? and price_band_id = ?";
+
+	private static final String[] COLUMN_NAMES = new String[] {"id", "forename"};
+
+
+	private Connection connection = mock();
+
+	private DataSource dataSource = mock();
+
+	private PreparedStatement preparedStatement = mock();
+
+	private ResultSet resultSet = mock();
+
+	private DatabaseMetaData databaseMetaData = mock();
+
+	private JdbcClient client = JdbcClient.create(dataSource);
+
+	private Map<String, Object> params = new HashMap<>();
+
+
+	@BeforeEach
+	public void setup() throws Exception {
+		given(dataSource.getConnection()).willReturn(connection);
+		given(connection.prepareStatement(anyString())).willReturn(preparedStatement);
+		given(preparedStatement.getConnection()).willReturn(connection);
+		given(preparedStatement.executeQuery()).willReturn(resultSet);
+		given(databaseMetaData.getDatabaseProductName()).willReturn("MySQL");
+		given(databaseMetaData.supportsBatchUpdates()).willReturn(true);
+	}
+
+
+	@Test
+	public void testQueryWithResultSetExtractor() throws SQLException {
+		given(resultSet.next()).willReturn(true);
+		given(resultSet.getInt("id")).willReturn(1);
+		given(resultSet.getString("forename")).willReturn("rod");
+
+		params.put("id", new SqlParameterValue(Types.DECIMAL, 1));
+		params.put("country", "UK");
+		Customer cust = client.sql(SELECT_NAMED_PARAMETERS).params(params).query(
+				rs -> {
+					rs.next();
+					Customer cust1 = new Customer();
+					cust1.setId(rs.getInt(COLUMN_NAMES[0]));
+					cust1.setForename(rs.getString(COLUMN_NAMES[1]));
+					return cust1;
+				});
+
+		assertThat(cust.getId()).as("Customer id was assigned correctly").isEqualTo(1);
+		assertThat(cust.getForename()).as("Customer forename was assigned correctly").isEqualTo("rod");
+		verify(connection).prepareStatement(SELECT_NAMED_PARAMETERS_PARSED);
+		verify(preparedStatement).setObject(1, 1, Types.DECIMAL);
+		verify(preparedStatement).setString(2, "UK");
+		verify(resultSet).close();
+		verify(preparedStatement).close();
+		verify(connection).close();
+	}
+
+	@Test
+	public void testQueryWithResultSetExtractorNoParameters() throws SQLException {
+		given(resultSet.next()).willReturn(true);
+		given(resultSet.getInt("id")).willReturn(1);
+		given(resultSet.getString("forename")).willReturn("rod");
+
+		Customer cust = client.sql(SELECT_NO_PARAMETERS).query(
+				rs -> {
+					rs.next();
+					Customer cust1 = new Customer();
+					cust1.setId(rs.getInt(COLUMN_NAMES[0]));
+					cust1.setForename(rs.getString(COLUMN_NAMES[1]));
+					return cust1;
+				});
+
+		assertThat(cust.getId()).as("Customer id was assigned correctly").isEqualTo(1);
+		assertThat(cust.getForename()).as("Customer forename was assigned correctly").isEqualTo("rod");
+		verify(connection).prepareStatement(SELECT_NO_PARAMETERS);
+		verify(resultSet).close();
+		verify(preparedStatement).close();
+		verify(connection).close();
+	}
+
+	@Test
+	public void testQueryWithRowCallbackHandler() throws SQLException {
+		given(resultSet.next()).willReturn(true, false);
+		given(resultSet.getInt("id")).willReturn(1);
+		given(resultSet.getString("forename")).willReturn("rod");
+
+		params.put("id", new SqlParameterValue(Types.DECIMAL, 1));
+		params.put("country", "UK");
+		final List<Customer> customers = new ArrayList<>();
+		client.sql(SELECT_NAMED_PARAMETERS).params(params).query(rs -> {
+			Customer cust = new Customer();
+			cust.setId(rs.getInt(COLUMN_NAMES[0]));
+			cust.setForename(rs.getString(COLUMN_NAMES[1]));
+			customers.add(cust);
+		});
+
+		assertThat(customers).hasSize(1);
+		assertThat(customers.get(0).getId()).as("Customer id was assigned correctly").isEqualTo(1);
+		assertThat(customers.get(0).getForename()).as("Customer forename was assigned correctly").isEqualTo("rod");
+		verify(connection).prepareStatement(SELECT_NAMED_PARAMETERS_PARSED);
+		verify(preparedStatement).setObject(1, 1, Types.DECIMAL);
+		verify(preparedStatement).setString(2, "UK");
+		verify(resultSet).close();
+		verify(preparedStatement).close();
+		verify(connection).close();
+	}
+
+	@Test
+	public void testQueryWithRowCallbackHandlerNoParameters() throws SQLException {
+		given(resultSet.next()).willReturn(true, false);
+		given(resultSet.getInt("id")).willReturn(1);
+		given(resultSet.getString("forename")).willReturn("rod");
+
+		final List<Customer> customers = new ArrayList<>();
+		client.sql(SELECT_NO_PARAMETERS).query(rs -> {
+			Customer cust = new Customer();
+			cust.setId(rs.getInt(COLUMN_NAMES[0]));
+			cust.setForename(rs.getString(COLUMN_NAMES[1]));
+			customers.add(cust);
+		});
+
+		assertThat(customers).hasSize(1);
+		assertThat(customers.get(0).getId()).as("Customer id was assigned correctly").isEqualTo(1);
+		assertThat(customers.get(0).getForename()).as("Customer forename was assigned correctly").isEqualTo("rod");
+		verify(connection).prepareStatement(SELECT_NO_PARAMETERS);
+		verify(resultSet).close();
+		verify(preparedStatement).close();
+		verify(connection).close();
+	}
+
+	@Test
+	public void testQueryWithRowMapper() throws SQLException {
+		given(resultSet.next()).willReturn(true, false);
+		given(resultSet.getInt("id")).willReturn(1);
+		given(resultSet.getString("forename")).willReturn("rod");
+
+		params.put("id", new SqlParameterValue(Types.DECIMAL, 1));
+		params.put("country", "UK");
+		List<Customer> customers = client.sql(SELECT_NAMED_PARAMETERS).params(params).query(
+				(rs, rownum) -> {
+					Customer cust = new Customer();
+					cust.setId(rs.getInt(COLUMN_NAMES[0]));
+					cust.setForename(rs.getString(COLUMN_NAMES[1]));
+					return cust;
+				}).list();
+
+		assertThat(customers).hasSize(1);
+		Customer cust = customers.get(0);
+		assertThat(cust.getId()).as("Customer id was assigned correctly").isEqualTo(1);
+		assertThat(cust.getForename()).as("Customer forename was assigned correctly").isEqualTo("rod");
+		verify(connection).prepareStatement(SELECT_NAMED_PARAMETERS_PARSED);
+		verify(preparedStatement).setObject(1, 1, Types.DECIMAL);
+		verify(preparedStatement).setString(2, "UK");
+		verify(resultSet).close();
+		verify(preparedStatement).close();
+		verify(connection).close();
+	}
+
+	@Test
+	public void testQueryWithRowMapperNoParameters() throws SQLException {
+		given(resultSet.next()).willReturn(true, false);
+		given(resultSet.getInt("id")).willReturn(1);
+		given(resultSet.getString("forename")).willReturn("rod");
+
+		Set<Customer> customers = client.sql(SELECT_NO_PARAMETERS).query(
+				(rs, rownum) -> {
+					Customer cust = new Customer();
+					cust.setId(rs.getInt(COLUMN_NAMES[0]));
+					cust.setForename(rs.getString(COLUMN_NAMES[1]));
+					return cust;
+				}).set();
+
+		assertThat(customers).hasSize(1);
+		Customer cust = customers.iterator().next();
+		assertThat(cust.getId()).as("Customer id was assigned correctly").isEqualTo(1);
+		assertThat(cust.getForename()).as("Customer forename was assigned correctly").isEqualTo("rod");
+		verify(connection).prepareStatement(SELECT_NO_PARAMETERS);
+		verify(resultSet).close();
+		verify(preparedStatement).close();
+		verify(connection).close();
+	}
+
+	@Test
+	public void testQueryForObjectWithRowMapper() throws SQLException {
+		given(resultSet.next()).willReturn(true, false);
+		given(resultSet.getInt("id")).willReturn(1);
+		given(resultSet.getString("forename")).willReturn("rod");
+
+		params.put("id", new SqlParameterValue(Types.DECIMAL, 1));
+		params.put("country", "UK");
+
+		Customer cust = client.sql(SELECT_NAMED_PARAMETERS).params(params).query(
+				(rs, rownum) -> {
+					Customer cust1 = new Customer();
+					cust1.setId(rs.getInt(COLUMN_NAMES[0]));
+					cust1.setForename(rs.getString(COLUMN_NAMES[1]));
+					return cust1;
+				}).single();
+
+		assertThat(cust.getId()).as("Customer id was assigned correctly").isEqualTo(1);
+		assertThat(cust.getForename()).as("Customer forename was assigned correctly").isEqualTo("rod");
+		verify(connection).prepareStatement(SELECT_NAMED_PARAMETERS_PARSED);
+		verify(preparedStatement).setObject(1, 1, Types.DECIMAL);
+		verify(preparedStatement).setString(2, "UK");
+		verify(resultSet).close();
+		verify(preparedStatement).close();
+		verify(connection).close();
+	}
+
+	@Test
+	public void testQueryForStreamWithRowMapper() throws SQLException {
+		given(resultSet.next()).willReturn(true, false);
+		given(resultSet.getInt("id")).willReturn(1);
+		given(resultSet.getString("forename")).willReturn("rod");
+
+		params.put("id", new SqlParameterValue(Types.DECIMAL, 1));
+		params.put("country", "UK");
+		AtomicInteger count = new AtomicInteger();
+
+		try (Stream<Customer> s = client.sql(SELECT_NAMED_PARAMETERS).params(params).query(
+				(rs, rownum) -> {
+					Customer cust1 = new Customer();
+					cust1.setId(rs.getInt(COLUMN_NAMES[0]));
+					cust1.setForename(rs.getString(COLUMN_NAMES[1]));
+					return cust1;
+				}).stream()) {
+			s.forEach(cust -> {
+				count.incrementAndGet();
+				assertThat(cust.getId()).as("Customer id was assigned correctly").isEqualTo(1);
+				assertThat(cust.getForename()).as("Customer forename was assigned correctly").isEqualTo("rod");
+			});
+		}
+
+		assertThat(count.get()).isEqualTo(1);
+		verify(connection).prepareStatement(SELECT_NAMED_PARAMETERS_PARSED);
+		verify(preparedStatement).setObject(1, 1, Types.DECIMAL);
+		verify(preparedStatement).setString(2, "UK");
+		verify(resultSet).close();
+		verify(preparedStatement).close();
+		verify(connection).close();
+	}
+
+	@Test
+	public void testUpdate() throws SQLException {
+		given(preparedStatement.executeUpdate()).willReturn(1);
+
+		params.put("perfId", 1);
+		params.put("priceId", 1);
+		int rowsAffected = client.sql(UPDATE_NAMED_PARAMETERS).params(params).update();
+
+		assertThat(rowsAffected).isEqualTo(1);
+		verify(connection).prepareStatement(UPDATE_NAMED_PARAMETERS_PARSED);
+		verify(preparedStatement).setObject(1, 1);
+		verify(preparedStatement).setObject(2, 1);
+		verify(preparedStatement).close();
+		verify(connection).close();
+	}
+
+	@Test
+	public void testUpdateWithTypedParameters() throws SQLException {
+		given(preparedStatement.executeUpdate()).willReturn(1);
+
+		params.put("perfId", new SqlParameterValue(Types.DECIMAL, 1));
+		params.put("priceId", new SqlParameterValue(Types.INTEGER, 1));
+		int rowsAffected = client.sql(UPDATE_NAMED_PARAMETERS).params(params).update();
+
+		assertThat(rowsAffected).isEqualTo(1);
+		verify(connection).prepareStatement(UPDATE_NAMED_PARAMETERS_PARSED);
+		verify(preparedStatement).setObject(1, 1, Types.DECIMAL);
+		verify(preparedStatement).setObject(2, 1, Types.INTEGER);
+		verify(preparedStatement).close();
+		verify(connection).close();
+	}
+
+}

spring-jdbc/src/test/java/org/springframework/jdbc/core/simple/JdbcClientQueryTests.java

@@ -0,0 +1,476 @@
+/*
+ * Copyright 2002-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.jdbc.core.simple;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.Types;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import javax.sql.DataSource;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+/**
+ * @author Juergen Hoeller
+ * @since 6.1
+ */
+public class JdbcClientQueryTests {
+
+	private Connection connection = mock();
+
+	private DataSource dataSource = mock();
+
+	private PreparedStatement preparedStatement = mock();
+
+	private ResultSet resultSet = mock();
+
+	private ResultSetMetaData resultSetMetaData = mock();
+
+	private JdbcClient client = JdbcClient.create(dataSource);
+
+
+	@BeforeEach
+	public void setup() throws Exception {
+		given(dataSource.getConnection()).willReturn(connection);
+		given(resultSetMetaData.getColumnCount()).willReturn(1);
+		given(resultSetMetaData.getColumnLabel(1)).willReturn("age");
+		given(connection.prepareStatement(anyString())).willReturn(preparedStatement);
+		given(preparedStatement.executeQuery()).willReturn(resultSet);
+	}
+
+	@AfterEach
+	public void verifyClose() throws Exception {
+		verify(preparedStatement).close();
+		verify(resultSet).close();
+		verify(connection).close();
+	}
+
+
+	// Indexed parameters
+
+	@Test
+	public void testQueryForListWithIndexedParam() throws Exception {
+		given(resultSet.getMetaData()).willReturn(resultSetMetaData);
+		given(resultSet.next()).willReturn(true, true, false);
+		given(resultSet.getObject(1)).willReturn(11, 12);
+
+		List<Map<String, Object>> li = client.sql("SELECT AGE FROM CUSTMR WHERE ID < ?")
+				.param(3).query().listOfRows();
+
+		assertThat(li.size()).as("All rows returned").isEqualTo(2);
+		assertThat(li.get(0).get("age")).as("First row is Integer").isEqualTo(11);
+		assertThat(li.get(1).get("age")).as("Second row is Integer").isEqualTo(12);
+
+		verify(connection).prepareStatement("SELECT AGE FROM CUSTMR WHERE ID < ?");
+		verify(preparedStatement).setObject(1, 3);
+	}
+
+	@Test
+	public void testQueryForListWithIndexedParamAndEmptyResult() throws Exception {
+		given(resultSet.next()).willReturn(false);
+
+		List<Map<String, Object>> li = client.sql("SELECT AGE FROM CUSTMR WHERE ID < ?")
+				.param(3).query().listOfRows();
+
+		assertThat(li.size()).as("All rows returned").isEqualTo(0);
+		verify(connection).prepareStatement("SELECT AGE FROM CUSTMR WHERE ID < ?");
+		verify(preparedStatement).setObject(1, 3);
+	}
+
+	@Test
+	public void testQueryForListWithIndexedParamAndSingleRowAndColumn() throws Exception {
+		given(resultSet.getMetaData()).willReturn(resultSetMetaData);
+		given(resultSet.next()).willReturn(true, false);
+		given(resultSet.getObject(1)).willReturn(11);
+
+		List<Map<String, Object>> li = client.sql("SELECT AGE FROM CUSTMR WHERE ID < ?")
+				.param(3).query().listOfRows();
+
+		assertThat(li.size()).as("All rows returned").isEqualTo(1);
+		assertThat(li.get(0).get("age")).as("First row is Integer").isEqualTo(11);
+		verify(connection).prepareStatement("SELECT AGE FROM CUSTMR WHERE ID < ?");
+		verify(preparedStatement).setObject(1, 3);
+	}
+
+	@Test
+	public void testQueryForListWithIndexedParamAndIntegerElementAndSingleRowAndColumn() throws Exception {
+		given(resultSet.getMetaData()).willReturn(resultSetMetaData);
+		given(resultSet.next()).willReturn(true, false);
+		given(resultSet.getInt(1)).willReturn(11);
+
+		List<Integer> li = client.sql("SELECT AGE FROM CUSTMR WHERE ID < ?")
+				.param(1, 3)
+				.query().singleColumn(Integer.class);
+
+		assertThat(li.size()).as("All rows returned").isEqualTo(1);
+		assertThat(li.get(0)).as("First row is Integer").isEqualTo(11);
+		verify(connection).prepareStatement("SELECT AGE FROM CUSTMR WHERE ID < ?");
+		verify(preparedStatement).setObject(1, 3);
+	}
+
+	@Test
+	public void testQueryForMapWithIndexedParamAndSingleRowAndColumn() throws Exception {
+		given(resultSet.getMetaData()).willReturn(resultSetMetaData);
+		given(resultSet.next()).willReturn(true, false);
+		given(resultSet.getObject(1)).willReturn(11);
+
+		Map<String, Object> map = client.sql("SELECT AGE FROM CUSTMR WHERE ID < ?")
+				.param(1, 3)
+				.query().singleRow();
+
+		assertThat(map.get("age")).as("Row is Integer").isEqualTo(11);
+		verify(connection).prepareStatement("SELECT AGE FROM CUSTMR WHERE ID < ?");
+		verify(preparedStatement).setObject(1, 3);
+	}
+
+	@Test
+	public void testQueryForObjectWithIndexedParamAndRowMapper() throws Exception {
+		given(resultSet.next()).willReturn(true, false);
+		given(resultSet.getInt(1)).willReturn(22);
+
+		Integer value = client.sql("SELECT AGE FROM CUSTMR WHERE ID = ?")
+				.param(1, 3)
+				.query((rs, rowNum) -> rs.getInt(1)).single();
+
+		assertThat(value).isEqualTo(22);
+		verify(connection).prepareStatement("SELECT AGE FROM CUSTMR WHERE ID = ?");
+		verify(preparedStatement).setObject(1, 3);
+	}
+
+	@Test
+	public void testQueryForOptionalWithIndexedParamAndRowMapper() throws Exception {
+		given(resultSet.next()).willReturn(true, false);
+		given(resultSet.getInt(1)).willReturn(22);
+
+		Optional<Integer> value = client.sql("SELECT AGE FROM CUSTMR WHERE ID = ?")
+				.param(1, 3)
+				.query((rs, rowNum) -> rs.getInt(1)).optional();
+
+		assertThat(value.get()).isEqualTo(22);
+		verify(connection).prepareStatement("SELECT AGE FROM CUSTMR WHERE ID = ?");
+		verify(preparedStatement).setObject(1, 3);
+	}
+
+	@Test
+	public void testQueryForObjectWithIndexedParamAndInteger() throws Exception {
+		given(resultSet.getMetaData()).willReturn(resultSetMetaData);
+		given(resultSet.next()).willReturn(true, false);
+		given(resultSet.getInt(1)).willReturn(22);
+
+		Integer value = client.sql("SELECT AGE FROM CUSTMR WHERE ID = ?")
+				.param(1, 3)
+				.query().singleValue(Integer.class);
+
+		assertThat(value).isEqualTo(22);
+		verify(connection).prepareStatement("SELECT AGE FROM CUSTMR WHERE ID = ?");
+		verify(preparedStatement).setObject(1, 3);
+	}
+
+	@Test
+	public void testQueryForIntWithIndexedParam() throws Exception {
+		given(resultSet.getMetaData()).willReturn(resultSetMetaData);
+		given(resultSet.next()).willReturn(true, false);
+		given(resultSet.getInt(1)).willReturn(22);
+
+		int i = client.sql("SELECT AGE FROM CUSTMR WHERE ID = ?")
+				.param(1, 3)
+				.query().singleValue(Integer.class);
+
+		assertThat(i).as("Return of an int").isEqualTo(22);
+		verify(connection).prepareStatement("SELECT AGE FROM CUSTMR WHERE ID = ?");
+		verify(preparedStatement).setObject(1, 3);
+	}
+
+
+	// Named parameters
+
+	@Test
+	public void testQueryForListWithNamedParam() throws Exception {
+		given(resultSet.getMetaData()).willReturn(resultSetMetaData);
+		given(resultSet.next()).willReturn(true, true, false);
+		given(resultSet.getObject(1)).willReturn(11, 12);
+
+		List<Map<String, Object>> li = client.sql("SELECT AGE FROM CUSTMR WHERE ID < :id")
+				.param("id", 3)
+				.query().listOfRows();
+
+		assertThat(li.size()).as("All rows returned").isEqualTo(2);
+		assertThat(li.get(0).get("age")).as("First row is Integer").isEqualTo(11);
+		assertThat(li.get(1).get("age")).as("Second row is Integer").isEqualTo(12);
+
+		verify(connection).prepareStatement("SELECT AGE FROM CUSTMR WHERE ID < ?");
+		verify(preparedStatement).setObject(1, 3);
+	}
+
+	@Test
+	public void testQueryForListWithNamedParamAndEmptyResult() throws Exception {
+		given(resultSet.next()).willReturn(false);
+
+		List<Map<String, Object>> li = client.sql("SELECT AGE FROM CUSTMR WHERE ID < :id")
+				.param("id", 3)
+				.query().listOfRows();
+
+		assertThat(li.size()).as("All rows returned").isEqualTo(0);
+		verify(connection).prepareStatement("SELECT AGE FROM CUSTMR WHERE ID < ?");
+		verify(preparedStatement).setObject(1, 3);
+	}
+
+	@Test
+	public void testQueryForListWithNamedParamAndSingleRowAndColumn() throws Exception {
+		given(resultSet.getMetaData()).willReturn(resultSetMetaData);
+		given(resultSet.next()).willReturn(true, false);
+		given(resultSet.getObject(1)).willReturn(11);
+
+		List<Map<String, Object>> li = client.sql("SELECT AGE FROM CUSTMR WHERE ID < :id")
+				.param("id", 3)
+				.query().listOfRows();
+
+		assertThat(li.size()).as("All rows returned").isEqualTo(1);
+		assertThat(li.get(0).get("age")).as("First row is Integer").isEqualTo(11);
+		verify(connection).prepareStatement("SELECT AGE FROM CUSTMR WHERE ID < ?");
+		verify(preparedStatement).setObject(1, 3);
+	}
+
+	@Test
+	public void testQueryForListWithNamedParamAndIntegerElementAndSingleRowAndColumn() throws Exception {
+		given(resultSet.getMetaData()).willReturn(resultSetMetaData);
+		given(resultSet.next()).willReturn(true, false);
+		given(resultSet.getInt(1)).willReturn(11);
+
+		List<Integer> li = client.sql("SELECT AGE FROM CUSTMR WHERE ID < :id")
+				.param("id", 3)
+				.query().singleColumn(Integer.class);
+
+		assertThat(li.size()).as("All rows returned").isEqualTo(1);
+		assertThat(li.get(0)).as("First row is Integer").isEqualTo(11);
+		verify(connection).prepareStatement("SELECT AGE FROM CUSTMR WHERE ID < ?");
+		verify(preparedStatement).setObject(1, 3);
+	}
+
+	@Test
+	public void testQueryForMapWithNamedParamAndSingleRowAndColumn() throws Exception {
+		given(resultSet.getMetaData()).willReturn(resultSetMetaData);
+		given(resultSet.next()).willReturn(true, false);
+		given(resultSet.getObject(1)).willReturn(11);
+
+		Map<String, Object> map = client.sql("SELECT AGE FROM CUSTMR WHERE ID < :id")
+				.param("id", 3)
+				.query().singleRow();
+
+		assertThat(map.get("age")).as("Row is Integer").isEqualTo(11);
+		verify(connection).prepareStatement("SELECT AGE FROM CUSTMR WHERE ID < ?");
+		verify(preparedStatement).setObject(1, 3);
+	}
+
+	@Test
+	public void testQueryForObjectWithNamedParamAndRowMapper() throws Exception {
+		given(resultSet.next()).willReturn(true, false);
+		given(resultSet.getInt(1)).willReturn(22);
+
+		Integer value = client.sql("SELECT AGE FROM CUSTMR WHERE ID = :id")
+				.param("id", 3)
+				.query((rs, rowNum) -> rs.getInt(1))
+				.single();
+
+		assertThat(value).isEqualTo(22);
+		verify(connection).prepareStatement("SELECT AGE FROM CUSTMR WHERE ID = ?");
+		verify(preparedStatement).setObject(1, 3);
+	}
+
+	@Test
+	public void testQueryForObjectWithNamedParamAndInteger() throws Exception {
+		given(resultSet.getMetaData()).willReturn(resultSetMetaData);
+		given(resultSet.next()).willReturn(true, false);
+		given(resultSet.getInt(1)).willReturn(22);
+
+		Integer value = client.sql("SELECT AGE FROM CUSTMR WHERE ID = :id")
+				.param("id", 3)
+				.query().singleValue(Integer.class);
+
+		assertThat(value).isEqualTo(22);
+		verify(connection).prepareStatement("SELECT AGE FROM CUSTMR WHERE ID = ?");
+		verify(preparedStatement).setObject(1, 3);
+	}
+
+	@Test
+	public void testQueryForObjectWithNamedParamAndList() throws Exception {
+		given(resultSet.getMetaData()).willReturn(resultSetMetaData);
+		given(resultSet.next()).willReturn(true, false);
+		given(resultSet.getInt(1)).willReturn(22);
+
+		Integer value = client.sql("SELECT AGE FROM CUSTMR WHERE ID IN (:ids)")
+				.param("ids", Arrays.asList(3, 4))
+				.query().singleValue(Integer.class);
+
+		assertThat(value).isEqualTo(22);
+		verify(connection).prepareStatement("SELECT AGE FROM CUSTMR WHERE ID IN (?, ?)");
+		verify(preparedStatement).setObject(1, 3);
+	}
+
+	@Test
+	public void testQueryForObjectWithNamedParamAndListOfExpressionLists() throws Exception {
+		given(resultSet.getMetaData()).willReturn(resultSetMetaData);
+		given(resultSet.next()).willReturn(true, false);
+		given(resultSet.getInt(1)).willReturn(22);
+
+		List<Object[]> l1 = new ArrayList<>();
+		l1.add(new Object[] {3, "Rod"});
+		l1.add(new Object[] {4, "Juergen"});
+		Integer value = client.sql("SELECT AGE FROM CUSTMR WHERE (ID, NAME) IN (:multiExpressionList)")
+				.param("multiExpressionList", l1)
+				.query().singleValue(Integer.class);
+
+		assertThat(value).isEqualTo(22);
+		verify(connection).prepareStatement("SELECT AGE FROM CUSTMR WHERE (ID, NAME) IN ((?, ?), (?, ?))");
+		verify(preparedStatement).setObject(1, 3);
+	}
+
+	@Test
+	public void testQueryForIntWithNamedParam() throws Exception {
+		given(resultSet.getMetaData()).willReturn(resultSetMetaData);
+		given(resultSet.next()).willReturn(true, false);
+		given(resultSet.getInt(1)).willReturn(22);
+
+		int i = client.sql("SELECT AGE FROM CUSTMR WHERE ID = :id")
+				.param("id", 3)
+				.query().singleValue(Integer.class);
+
+		assertThat(i).as("Return of an int").isEqualTo(22);
+		verify(connection).prepareStatement("SELECT AGE FROM CUSTMR WHERE ID = ?");
+		verify(preparedStatement).setObject(1, 3);
+	}
+
+	@Test
+	public void testQueryForLongWithParamBean() throws Exception {
+		given(resultSet.getMetaData()).willReturn(resultSetMetaData);
+		given(resultSet.next()).willReturn(true, false);
+		given(resultSet.getLong(1)).willReturn(87L);
+
+		long l = client.sql("SELECT AGE FROM CUSTMR WHERE ID = :id")
+				.paramSource(new ParameterBean(3))
+				.query().singleValue(Long.class);
+
+		assertThat(l).as("Return of a long").isEqualTo(87);
+		verify(connection).prepareStatement("SELECT AGE FROM CUSTMR WHERE ID = ?");
+		verify(preparedStatement).setObject(1, 3, Types.INTEGER);
+	}
+
+	@Test
+	public void testQueryForLongWithParamBeanWithCollection() throws Exception {
+		given(resultSet.getMetaData()).willReturn(resultSetMetaData);
+		given(resultSet.next()).willReturn(true, false);
+		given(resultSet.getLong(1)).willReturn(87L);
+
+		long l = client.sql("SELECT AGE FROM CUSTMR WHERE ID IN (:ids)")
+				.paramSource(new ParameterCollectionBean(3, 5))
+				.query().singleValue(Long.class);
+
+		assertThat(l).as("Return of a long").isEqualTo(87);
+		verify(connection).prepareStatement("SELECT AGE FROM CUSTMR WHERE ID IN (?, ?)");
+		verify(preparedStatement).setObject(1, 3);
+		verify(preparedStatement).setObject(2, 5);
+	}
+
+	@Test
+	public void testQueryForLongWithParamRecord() throws Exception {
+		given(resultSet.getMetaData()).willReturn(resultSetMetaData);
+		given(resultSet.next()).willReturn(true, false);
+		given(resultSet.getLong(1)).willReturn(87L);
+
+		long l = client.sql("SELECT AGE FROM CUSTMR WHERE ID = :id")
+				.paramSource(new ParameterRecord(3))
+				.query().singleValue(Long.class);
+
+		assertThat(l).as("Return of a long").isEqualTo(87);
+		verify(connection).prepareStatement("SELECT AGE FROM CUSTMR WHERE ID = ?");
+		verify(preparedStatement).setObject(1, 3, Types.INTEGER);
+	}
+
+	@Test
+	public void testQueryForLongWithParamFieldHolder() throws Exception {
+		given(resultSet.getMetaData()).willReturn(resultSetMetaData);
+		given(resultSet.next()).willReturn(true, false);
+		given(resultSet.getLong(1)).willReturn(87L);
+
+		long l = client.sql("SELECT AGE FROM CUSTMR WHERE ID = :id")
+				.paramSource(new ParameterFieldHolder(3))
+				.query().singleValue(Long.class);
+
+		assertThat(l).as("Return of a long").isEqualTo(87);
+		verify(connection).prepareStatement("SELECT AGE FROM CUSTMR WHERE ID = ?");
+		verify(preparedStatement).setObject(1, 3, Types.INTEGER);
+	}
+
+
+	static class ParameterBean {
+
+		private final int id;
+
+		public ParameterBean(int id) {
+			this.id = id;
+		}
+
+		public int getId() {
+			return id;
+		}
+	}
+
+
+	static class ParameterCollectionBean {
+
+		private final Collection<Integer> ids;
+
+		public ParameterCollectionBean(Integer... ids) {
+			this.ids = Arrays.asList(ids);
+		}
+
+		public Collection<Integer> getIds() {
+			return ids;
+		}
+	}
+
+
+	record ParameterRecord(int id) {
+	}
+
+
+	static class ParameterFieldHolder {
+
+		public ParameterFieldHolder(int id) {
+			this.id = id;
+		}
+
+		public int id;
+	}
+
+}