Introduce queryForStream on JdbcTemplate and NamedParameterJdbcTemplate

Closes gh-18474

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

@@ -19,6 +19,7 @@ package org.springframework.jdbc.core;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Stream;
 
 import org.springframework.dao.DataAccessException;
 import org.springframework.dao.IncorrectResultSizeDataAccessException;
@@ -100,7 +101,7 @@ public interface JdbcOperations {
 	 * @param rse a callback that will extract all rows of results
 	 * @return an arbitrary result object, as returned by the ResultSetExtractor
 	 * @throws DataAccessException if there is any problem executing the query
-	 * @see #query(String, Object[], ResultSetExtractor)
+	 * @see #query(String, ResultSetExtractor, Object...)
 	 */
 	@Nullable
 	<T> T query(String sql, ResultSetExtractor<T> rse) throws DataAccessException;
@@ -114,7 +115,7 @@ public interface JdbcOperations {
 	 * @param sql the SQL query to execute
 	 * @param rch a callback that will extract results, one row at a time
 	 * @throws DataAccessException if there is any problem executing the query
-	 * @see #query(String, Object[], RowCallbackHandler)
+	 * @see #query(String, RowCallbackHandler, Object...)
 	 */
 	void query(String sql, RowCallbackHandler rch) throws DataAccessException;
 
@@ -128,10 +129,26 @@ public interface JdbcOperations {
 	 * @param rowMapper a callback that will map one object per row
 	 * @return the result List, containing mapped objects
 	 * @throws DataAccessException if there is any problem executing the query
-	 * @see #query(String, Object[], RowMapper)
+	 * @see #query(String, RowMapper, Object...)
 	 */
 	<T> List<T> query(String sql, RowMapper<T> rowMapper) throws DataAccessException;
 
+	/**
+	 * Execute a query given static SQL, mapping each row to a result object
+	 * via a RowMapper, and turning it into an iterable and closeable Stream.
+	 * <p>Uses a JDBC Statement, not a PreparedStatement. If you want to
+	 * execute a static query with a PreparedStatement, use the overloaded
+	 * {@code query} method with {@code null} as argument array.
+	 * @param sql the SQL query to execute
+	 * @param rowMapper a callback that will map one object per row
+	 * @return the result Stream, containing mapped objects, needing to be
+	 * closed once fully processed (e.g. through a try-with-resources clause)
+	 * @throws DataAccessException if there is any problem executing the query
+	 * @since 5.3
+	 * @see #queryForStream(String, RowMapper, Object...)
+	 */
+	<T> Stream<T> queryForStream(String sql, RowMapper<T> rowMapper) throws DataAccessException;
+
 	/**
 	 * Execute a query given static SQL, mapping a single result row to a
 	 * result object via a RowMapper.
@@ -146,7 +163,7 @@ public interface JdbcOperations {
 	 * @throws IncorrectResultSizeDataAccessException if the query does not
 	 * return exactly one row
 	 * @throws DataAccessException if there is any problem executing the query
-	 * @see #queryForObject(String, Object[], RowMapper)
+	 * @see #queryForObject(String, RowMapper, Object...)
 	 */
 	@Nullable
 	<T> T queryForObject(String sql, RowMapper<T> rowMapper) throws DataAccessException;
@@ -166,7 +183,7 @@ public interface JdbcOperations {
 	 * @throws IncorrectResultSizeDataAccessException if the query does not return
 	 * exactly one row, or does not return exactly one column in that row
 	 * @throws DataAccessException if there is any problem executing the query
-	 * @see #queryForObject(String, Object[], Class)
+	 * @see #queryForObject(String, Class, Object...)
 	 */
 	@Nullable
 	<T> T queryForObject(String sql, Class<T> requiredType) throws DataAccessException;
@@ -184,7 +201,7 @@ public interface JdbcOperations {
 	 * @throws IncorrectResultSizeDataAccessException if the query does not
 	 * return exactly one row
 	 * @throws DataAccessException if there is any problem executing the query
-	 * @see #queryForMap(String, Object[])
+	 * @see #queryForMap(String, Object...)
 	 * @see ColumnMapRowMapper
 	 */
 	Map<String, Object> queryForMap(String sql) throws DataAccessException;
@@ -201,7 +218,7 @@ public interface JdbcOperations {
 	 * (for example, {@code Integer.class})
 	 * @return a List of objects that match the specified element type
 	 * @throws DataAccessException if there is any problem executing the query
-	 * @see #queryForList(String, Object[], Class)
+	 * @see #queryForList(String, Class, Object...)
 	 * @see SingleColumnRowMapper
 	 */
 	<T> List<T> queryForList(String sql, Class<T> elementType) throws DataAccessException;
@@ -218,7 +235,7 @@ public interface JdbcOperations {
 	 * @param sql the SQL query to execute
 	 * @return an List that contains a Map per row
 	 * @throws DataAccessException if there is any problem executing the query
-	 * @see #queryForList(String, Object[])
+	 * @see #queryForList(String, Object...)
 	 */
 	List<Map<String, Object>> queryForList(String sql) throws DataAccessException;
 
@@ -237,7 +254,7 @@ public interface JdbcOperations {
 	 * @return an SqlRowSet representation (possibly a wrapper around a
 	 * {@code javax.sql.rowset.CachedRowSet})
 	 * @throws DataAccessException if there is any problem executing the query
-	 * @see #queryForRowSet(String, Object[])
+	 * @see #queryForRowSet(String, Object...)
 	 * @see SqlRowSetResultSetExtractor
 	 * @see javax.sql.rowset.CachedRowSet
 	 */
@@ -323,7 +340,8 @@ public interface JdbcOperations {
 	 * @throws DataAccessException if there is any problem
 	 */
 	@Nullable
-	<T> T query(String sql, @Nullable PreparedStatementSetter pss, ResultSetExtractor<T> rse) throws DataAccessException;
+	<T> T query(String sql, @Nullable PreparedStatementSetter pss, ResultSetExtractor<T> rse)
+			throws DataAccessException;
 
 	/**
 	 * Query given SQL to create a prepared statement from SQL and a list of arguments
@@ -466,7 +484,8 @@ public interface JdbcOperations {
 	 * @return the result List, containing mapped objects
 	 * @throws DataAccessException if the query fails
 	 */
-	<T> List<T> query(String sql, @Nullable PreparedStatementSetter pss, RowMapper<T> rowMapper) throws DataAccessException;
+	<T> List<T> query(String sql, @Nullable PreparedStatementSetter pss, RowMapper<T> rowMapper)
+			throws DataAccessException;
 
 	/**
 	 * Query given SQL to create a prepared statement from SQL and a list of
@@ -514,6 +533,58 @@ public interface JdbcOperations {
 	 */
 	<T> List<T> query(String sql, RowMapper<T> rowMapper, @Nullable Object... args) throws DataAccessException;
 
+	/**
+	 * Query using a prepared statement, mapping each row to a result object
+	 * via a RowMapper, and turning it into an iterable and closeable Stream.
+	 * <p>A PreparedStatementCreator can either be implemented directly or
+	 * configured through a PreparedStatementCreatorFactory.
+	 * @param psc a callback that creates a PreparedStatement given a Connection
+	 * @param rowMapper a callback that will map one object per row
+	 * @return the result Stream, containing mapped objects, needing to be
+	 * closed once fully processed (e.g. through a try-with-resources clause)
+	 * @throws DataAccessException if there is any problem
+	 * @see PreparedStatementCreatorFactory
+	 * @since 5.3
+	 */
+	<T> Stream<T> queryForStream(PreparedStatementCreator psc, RowMapper<T> rowMapper) throws DataAccessException;
+
+	/**
+	 * Query given SQL to create a prepared statement from SQL and a
+	 * PreparedStatementSetter implementation that knows how to bind values
+	 * to the query, mapping each row to a result object via a RowMapper,
+	 * and turning it into an iterable and closeable Stream.
+	 * @param sql the SQL query to execute
+	 * @param pss a callback that knows how to set values on the prepared statement.
+	 * If this is {@code null}, the SQL will be assumed to contain no bind parameters.
+	 * Even if there are no bind parameters, this callback may be used to set the
+	 * fetch size and other performance options.
+	 * @param rowMapper a callback that will map one object per row
+	 * @return the result Stream, containing mapped objects, needing to be
+	 * closed once fully processed (e.g. through a try-with-resources clause)
+	 * @throws DataAccessException if the query fails
+	 * @since 5.3
+	 */
+	<T> Stream<T> queryForStream(String sql, @Nullable PreparedStatementSetter pss, RowMapper<T> rowMapper)
+			throws DataAccessException;
+
+	/**
+	 * Query given SQL to create a prepared statement from SQL and a list of
+	 * arguments to bind to the query, mapping each row to a result object
+	 * via a RowMapper, and turning it into an iterable and closeable Stream.
+	 * @param sql the SQL query to execute
+	 * @param rowMapper a callback that will map one object per row
+	 * @param args arguments to bind to the query
+	 * (leaving it to the PreparedStatement to guess the corresponding SQL type);
+	 * may also contain {@link SqlParameterValue} objects which indicate not
+	 * only the argument value but also the SQL type and optionally the scale
+	 * @return the result Stream, containing mapped objects, needing to be
+	 * closed once fully processed (e.g. through a try-with-resources clause)
+	 * @throws DataAccessException if the query fails
+	 * @since 5.3
+	 */
+	<T> Stream<T> queryForStream(String sql, RowMapper<T> rowMapper, @Nullable Object... args)
+			throws DataAccessException;
+
 	/**
 	 * Query given SQL to create a prepared statement from SQL and a list
 	 * of arguments to bind to the query, mapping a single result row to a
@@ -903,6 +974,7 @@ public interface JdbcOperations {
 	 * @param sql the SQL statement to execute
 	 * @param batchArgs the List of Object arrays containing the batch of arguments for the query
 	 * @return an array containing the numbers of rows affected by each update in the batch
+	 * @throws DataAccessException if there is any problem issuing the update
 	 */
 	int[] batchUpdate(String sql, List<Object[]> batchArgs) throws DataAccessException;
 
@@ -913,6 +985,7 @@ public interface JdbcOperations {
 	 * @param argTypes the SQL types of the arguments
 	 * (constants from {@code java.sql.Types})
 	 * @return an array containing the numbers of rows affected by each update in the batch
+	 * @throws DataAccessException if there is any problem issuing the update
 	 */
 	int[] batchUpdate(String sql, List<Object[]> batchArgs, int[] argTypes) throws DataAccessException;
 
@@ -926,6 +999,7 @@ public interface JdbcOperations {
 	 * @param pss the ParameterizedPreparedStatementSetter to use
 	 * @return an array containing for each batch another array containing the numbers of rows affected
 	 * by each update in the batch
+	 * @throws DataAccessException if there is any problem issuing the update
 	 * @since 3.1
 	 */
 	<T> int[][] batchUpdate(String sql, Collection<T> batchArgs, int batchSize,

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

@@ -34,12 +34,17 @@ import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Spliterator;
+import java.util.function.Consumer;
+import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
 
 import javax.sql.DataSource;
 
 import org.springframework.dao.DataAccessException;
 import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.dao.support.DataAccessUtils;
+import org.springframework.jdbc.InvalidResultSetAccessException;
 import org.springframework.jdbc.SQLWarningException;
 import org.springframework.jdbc.UncategorizedSQLException;
 import org.springframework.jdbc.datasource.ConnectionProxy;
@@ -363,9 +368,8 @@ public class JdbcTemplate extends JdbcAccessor implements JdbcOperations {
 	// Methods dealing with static SQL (java.sql.Statement)
 	//-------------------------------------------------------------------------
 
-	@Override
 	@Nullable
-	public <T> T execute(StatementCallback<T> action) throws DataAccessException {
+	private <T> T execute(StatementCallback<T> action, boolean closeResources) throws DataAccessException {
 		Assert.notNull(action, "Callback object must not be null");
 
 		Connection con = DataSourceUtils.getConnection(obtainDataSource());
@@ -388,11 +392,19 @@ public class JdbcTemplate extends JdbcAccessor implements JdbcOperations {
 			throw translateException("StatementCallback", sql, ex);
 		}
 		finally {
-			JdbcUtils.closeStatement(stmt);
+			if (closeResources) {
-			DataSourceUtils.releaseConnection(con, getDataSource());
+				JdbcUtils.closeStatement(stmt);
+				DataSourceUtils.releaseConnection(con, getDataSource());
+			}
 		}
 	}
 
+	@Override
+	@Nullable
+	public <T> T execute(StatementCallback<T> action) throws DataAccessException {
+		return execute(action, true);
+	}
+
 	@Override
 	public void execute(final String sql) throws DataAccessException {
 		if (logger.isDebugEnabled()) {
@@ -415,7 +427,7 @@ public class JdbcTemplate extends JdbcAccessor implements JdbcOperations {
 			}
 		}
 
-		execute(new ExecuteStatementCallback());
+		execute(new ExecuteStatementCallback(), true);
 	}
 
 	@Override
@@ -449,7 +461,7 @@ public class JdbcTemplate extends JdbcAccessor implements JdbcOperations {
 			}
 		}
 
-		return execute(new QueryStatementCallback());
+		return execute(new QueryStatementCallback(), true);
 	}
 
 	@Override
@@ -462,6 +474,28 @@ public class JdbcTemplate extends JdbcAccessor implements JdbcOperations {
 		return result(query(sql, new RowMapperResultSetExtractor<>(rowMapper)));
 	}
 
+	@Override
+	public <T> Stream<T> queryForStream(String sql, RowMapper<T> rowMapper) throws DataAccessException {
+		class StreamStatementCallback implements StatementCallback<Stream<T>>, SqlProvider {
+			@Override
+			public Stream<T> doInStatement(Statement stmt) throws SQLException {
+				ResultSet rs = stmt.executeQuery(sql);
+				Connection con = stmt.getConnection();
+				return new ResultSetSpliterator<>(rs, rowMapper).stream().onClose(() -> {
+					JdbcUtils.closeResultSet(rs);
+					JdbcUtils.closeStatement(stmt);
+					DataSourceUtils.releaseConnection(con, getDataSource());
+				});
+			}
+			@Override
+			public String getSql() {
+				return sql;
+			}
+		}
+
+		return result(execute(new StreamStatementCallback(), false));
+	}
+
 	@Override
 	public Map<String, Object> queryForMap(String sql) throws DataAccessException {
 		return result(queryForObject(sql, getColumnMapRowMapper()));
@@ -520,7 +554,7 @@ public class JdbcTemplate extends JdbcAccessor implements JdbcOperations {
 			}
 		}
 
-		return updateCount(execute(new UpdateStatementCallback()));
+		return updateCount(execute(new UpdateStatementCallback(), true));
 	}
 
 	@Override
@@ -587,7 +621,7 @@ public class JdbcTemplate extends JdbcAccessor implements JdbcOperations {
 			}
 		}
 
-		int[] result = execute(new BatchUpdateStatementCallback());
+		int[] result = execute(new BatchUpdateStatementCallback(), true);
 		Assert.state(result != null, "No update counts");
 		return result;
 	}
@@ -597,9 +631,8 @@ public class JdbcTemplate extends JdbcAccessor implements JdbcOperations {
 	// Methods dealing with prepared statements
 	//-------------------------------------------------------------------------
 
-	@Override
 	@Nullable
-	public <T> T execute(PreparedStatementCreator psc, PreparedStatementCallback<T> action)
+	private <T> T execute(PreparedStatementCreator psc, PreparedStatementCallback<T> action, boolean closeResources)
 			throws DataAccessException {
 
 		Assert.notNull(psc, "PreparedStatementCreator must not be null");
@@ -633,18 +666,28 @@ public class JdbcTemplate extends JdbcAccessor implements JdbcOperations {
 			throw translateException("PreparedStatementCallback", sql, ex);
 		}
 		finally {
-			if (psc instanceof ParameterDisposer) {
+			if (closeResources) {
-				((ParameterDisposer) psc).cleanupParameters();
+				if (psc instanceof ParameterDisposer) {
+					((ParameterDisposer) psc).cleanupParameters();
+				}
+				JdbcUtils.closeStatement(ps);
+				DataSourceUtils.releaseConnection(con, getDataSource());
 			}
-			JdbcUtils.closeStatement(ps);
-			DataSourceUtils.releaseConnection(con, getDataSource());
 		}
 	}
 
+	@Override
+	@Nullable
+	public <T> T execute(PreparedStatementCreator psc, PreparedStatementCallback<T> action)
+			throws DataAccessException {
+
+		return execute(psc, action, true);
+	}
+
 	@Override
 	@Nullable
 	public <T> T execute(String sql, PreparedStatementCallback<T> action) throws DataAccessException {
-		return execute(new SimplePreparedStatementCreator(sql), action);
+		return execute(new SimplePreparedStatementCreator(sql), action, true);
 	}
 
 	/**
@@ -685,7 +728,7 @@ public class JdbcTemplate extends JdbcAccessor implements JdbcOperations {
 					}
 				}
 			}
-		});
+		}, true);
 	}
 
 	@Override
@@ -768,6 +811,54 @@ public class JdbcTemplate extends JdbcAccessor implements JdbcOperations {
 		return result(query(sql, args, new RowMapperResultSetExtractor<>(rowMapper)));
 	}
 
+	/**
+	 * Query using a prepared statement, allowing for a PreparedStatementCreator
+	 * and a PreparedStatementSetter. Most other query methods use this method,
+	 * but application code will always work with either a creator or a setter.
+	 * @param psc a callback that creates a PreparedStatement given a Connection
+	 * @param pss a callback that knows how to set values on the prepared statement.
+	 * If this is {@code null}, the SQL will be assumed to contain no bind parameters.
+	 * @param rowMapper a callback that will map one object per row
+	 * @return the result Stream, containing mapped objects, needing to be
+	 * closed once fully processed (e.g. through a try-with-resources clause)
+	 * @throws DataAccessException if the query fails
+	 * @since 5.3
+	 */
+	public <T> Stream<T> queryForStream(PreparedStatementCreator psc, @Nullable PreparedStatementSetter pss,
+			RowMapper<T> rowMapper) throws DataAccessException {
+
+		return result(execute(psc, ps -> {
+			if (pss != null) {
+				pss.setValues(ps);
+			}
+			ResultSet rs = ps.executeQuery();
+			Connection con = ps.getConnection();
+			return new ResultSetSpliterator<>(rs, rowMapper).stream().onClose(() -> {
+				JdbcUtils.closeResultSet(rs);
+				if (pss instanceof ParameterDisposer) {
+					((ParameterDisposer) pss).cleanupParameters();
+				}
+				JdbcUtils.closeStatement(ps);
+				DataSourceUtils.releaseConnection(con, getDataSource());
+			});
+		}, false));
+	}
+
+	@Override
+	public <T> Stream<T> queryForStream(PreparedStatementCreator psc, RowMapper<T> rowMapper) throws DataAccessException {
+		return queryForStream(psc, null, rowMapper);
+	}
+
+	@Override
+	public <T> Stream<T> queryForStream(String sql, @Nullable PreparedStatementSetter pss, RowMapper<T> rowMapper) throws DataAccessException {
+		return queryForStream(new SimplePreparedStatementCreator(sql), pss, rowMapper);
+	}
+
+	@Override
+	public <T> Stream<T> queryForStream(String sql, RowMapper<T> rowMapper, @Nullable Object... args) throws DataAccessException {
+		return queryForStream(new SimplePreparedStatementCreator(sql), newArgPreparedStatementSetter(args), rowMapper);
+	}
+
 	@Override
 	@Nullable
 	public <T> T queryForObject(String sql, Object[] args, int[] argTypes, RowMapper<T> rowMapper)
@@ -875,7 +966,7 @@ public class JdbcTemplate extends JdbcAccessor implements JdbcOperations {
 					((ParameterDisposer) pss).cleanupParameters();
 				}
 			}
-		}));
+		}, true));
 	}
 
 	@Override
@@ -909,7 +1000,7 @@ public class JdbcTemplate extends JdbcAccessor implements JdbcOperations {
 				logger.trace("SQL update affected " + rows + " rows and returned " + generatedKeys.size() + " keys");
 			}
 			return rows;
-		}));
+		}, true));
 	}
 
 	@Override
@@ -1610,4 +1701,57 @@ public class JdbcTemplate extends JdbcAccessor implements JdbcOperations {
 		}
 	}
 
+
+	/**
+	 * Spliterator for queryForStream adaptation of a ResultSet to a Stream.
+	 * @since 5.3
+	 */
+	private static class ResultSetSpliterator<T> implements Spliterator<T> {
+
+		private final ResultSet rs;
+
+		private final RowMapper<T> rowMapper;
+
+		private int rowNum = 0;
+
+		public ResultSetSpliterator(ResultSet rs, RowMapper<T> rowMapper) {
+			this.rs = rs;
+			this.rowMapper = rowMapper;
+		}
+
+		@Override
+		public boolean tryAdvance(Consumer<? super T> action) {
+			try {
+				if (this.rs.next()) {
+					action.accept(this.rowMapper.mapRow(this.rs, this.rowNum++));
+					return true;
+				}
+				return false;
+			}
+			catch (SQLException ex) {
+				throw new InvalidResultSetAccessException(ex);
+			}
+		}
+
+		@Override
+		@Nullable
+		public Spliterator<T> trySplit() {
+			return null;
+		}
+
+		@Override
+		public long estimateSize() {
+			return Long.MAX_VALUE;
+		}
+
+		@Override
+		public int characteristics() {
+			return Spliterator.ORDERED;
+		}
+
+		public Stream<T> stream() {
+			return StreamSupport.stream(this, false);
+		}
+	}
+
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2018 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -18,6 +18,7 @@ package org.springframework.jdbc.core.namedparam;
 
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Stream;
 
 import org.springframework.dao.DataAccessException;
 import org.springframework.jdbc.core.JdbcOperations;
@@ -130,7 +131,7 @@ public interface NamedParameterJdbcOperations {
 	 * (leaving it to the PreparedStatement to guess the corresponding SQL type)
 	 * @param rse object that will extract results
 	 * @return an arbitrary result object, as returned by the ResultSetExtractor
-	 * @throws org.springframework.dao.DataAccessException if the query fails
+	 * @throws DataAccessException if the query fails
 	 */
 	@Nullable
 	<T> T query(String sql, Map<String, ?> paramMap, ResultSetExtractor<T> rse)
@@ -145,7 +146,7 @@ public interface NamedParameterJdbcOperations {
 	 * @param sql the SQL query to execute
 	 * @param rse object that will extract results
 	 * @return an arbitrary result object, as returned by the ResultSetExtractor
-	 * @throws org.springframework.dao.DataAccessException if the query fails
+	 * @throws DataAccessException if the query fails
 	 */
 	@Nullable
 	<T> T query(String sql, ResultSetExtractor<T> rse) throws DataAccessException;
@@ -170,7 +171,7 @@ public interface NamedParameterJdbcOperations {
 	 * @param paramMap map of parameters to bind to the query
 	 * (leaving it to the PreparedStatement to guess the corresponding SQL type)
 	 * @param rch object that will extract results, one row at a time
-	 * @throws org.springframework.dao.DataAccessException if the query fails
+	 * @throws DataAccessException if the query fails
 	 */
 	void query(String sql, Map<String, ?> paramMap, RowCallbackHandler rch) throws DataAccessException;
 
@@ -182,7 +183,7 @@ public interface NamedParameterJdbcOperations {
 	 * equivalent to a query call with an empty parameter Map.
 	 * @param sql the SQL query to execute
 	 * @param rch object that will extract results, one row at a time
-	 * @throws org.springframework.dao.DataAccessException if the query fails
+	 * @throws DataAccessException if the query fails
 	 */
 	void query(String sql, RowCallbackHandler rch) throws DataAccessException;
 
@@ -194,7 +195,7 @@ public interface NamedParameterJdbcOperations {
 	 * @param paramSource container of arguments to bind to the query
 	 * @param rowMapper object that will map one object per row
 	 * @return the result List, containing mapped objects
-	 * @throws org.springframework.dao.DataAccessException if the query fails
+	 * @throws DataAccessException if the query fails
 	 */
 	<T> List<T> query(String sql, SqlParameterSource paramSource, RowMapper<T> rowMapper)
 			throws DataAccessException;
@@ -208,7 +209,7 @@ public interface NamedParameterJdbcOperations {
 	 * (leaving it to the PreparedStatement to guess the corresponding SQL type)
 	 * @param rowMapper object that will map one object per row
 	 * @return the result List, containing mapped objects
-	 * @throws org.springframework.dao.DataAccessException if the query fails
+	 * @throws DataAccessException if the query fails
 	 */
 	<T> List<T> query(String sql, Map<String, ?> paramMap, RowMapper<T> rowMapper)
 			throws DataAccessException;
@@ -222,10 +223,41 @@ public interface NamedParameterJdbcOperations {
 	 * @param sql the SQL query to execute
 	 * @param rowMapper object that will map one object per row
 	 * @return the result List, containing mapped objects
-	 * @throws org.springframework.dao.DataAccessException if the query fails
+	 * @throws DataAccessException if the query fails
 	 */
 	<T> List<T> query(String sql, RowMapper<T> rowMapper) throws DataAccessException;
 
+	/**
+	 * Query given SQL to create a prepared statement from SQL and a list
+	 * of arguments to bind to the query, mapping each row to a Java object
+	 * via a RowMapper, and turning it into an iterable and closeable Stream.
+	 * @param sql the SQL query to execute
+	 * @param paramSource container of arguments to bind to the query
+	 * @param rowMapper object that will map one object per row
+	 * @return the result Stream, containing mapped objects, needing to be
+	 * closed once fully processed (e.g. through a try-with-resources clause)
+	 * @throws DataAccessException if the query fails
+	 * @since 5.3
+	 */
+	<T> Stream<T> queryForStream(String sql, SqlParameterSource paramSource, RowMapper<T> rowMapper)
+			throws DataAccessException;
+
+	/**
+	 * Query given SQL to create a prepared statement from SQL and a list
+	 * of arguments to bind to the query, mapping each row to a Java object
+	 * via a RowMapper, and turning it into an iterable and closeable Stream.
+	 * @param sql the SQL query to execute
+	 * @param paramMap map of parameters to bind to the query
+	 * (leaving it to the PreparedStatement to guess the corresponding SQL type)
+	 * @param rowMapper object that will map one object per row
+	 * @return the result Stream, containing mapped objects, needing to be
+	 * closed once fully processed (e.g. through a try-with-resources clause)
+	 * @throws DataAccessException if the query fails
+	 * @since 5.3
+	 */
+	<T> Stream<T> queryForStream(String sql, Map<String, ?> paramMap, RowMapper<T> rowMapper)
+			throws DataAccessException;
+
 	/**
 	 * Query given SQL to create a prepared statement from SQL and a list
 	 * of arguments to bind to the query, mapping a single result row to a
@@ -238,7 +270,7 @@ public interface NamedParameterJdbcOperations {
 	 * @throws org.springframework.dao.IncorrectResultSizeDataAccessException
 	 * if the query does not return exactly one row, or does not return exactly
 	 * one column in that row
-	 * @throws org.springframework.dao.DataAccessException if the query fails
+	 * @throws DataAccessException if the query fails
 	 */
 	@Nullable
 	<T> T queryForObject(String sql, SqlParameterSource paramSource, RowMapper<T> rowMapper)
@@ -257,7 +289,7 @@ public interface NamedParameterJdbcOperations {
 	 * @throws org.springframework.dao.IncorrectResultSizeDataAccessException
 	 * if the query does not return exactly one row, or does not return exactly
 	 * one column in that row
-	 * @throws org.springframework.dao.DataAccessException if the query fails
+	 * @throws DataAccessException if the query fails
 	 */
 	@Nullable
 	<T> T queryForObject(String sql, Map<String, ?> paramMap, RowMapper<T> rowMapper)
@@ -275,7 +307,7 @@ public interface NamedParameterJdbcOperations {
 	 * @throws org.springframework.dao.IncorrectResultSizeDataAccessException
 	 * if the query does not return exactly one row, or does not return exactly
 	 * one column in that row
-	 * @throws org.springframework.dao.DataAccessException if the query fails
+	 * @throws DataAccessException if the query fails
 	 * @see org.springframework.jdbc.core.JdbcTemplate#queryForObject(String, Class)
 	 */
 	@Nullable
@@ -295,7 +327,7 @@ public interface NamedParameterJdbcOperations {
 	 * @throws org.springframework.dao.IncorrectResultSizeDataAccessException
 	 * if the query does not return exactly one row, or does not return exactly
 	 * one column in that row
-	 * @throws org.springframework.dao.DataAccessException if the query fails
+	 * @throws DataAccessException if the query fails
 	 * @see org.springframework.jdbc.core.JdbcTemplate#queryForObject(String, Class)
 	 */
 	@Nullable
@@ -312,7 +344,7 @@ public interface NamedParameterJdbcOperations {
 	 * @return the result Map (one entry for each column, using the column name as the key)
 	 * @throws org.springframework.dao.IncorrectResultSizeDataAccessException
 	 * if the query does not return exactly one row
-	 * @throws org.springframework.dao.DataAccessException if the query fails
+	 * @throws DataAccessException if the query fails
 	 * @see org.springframework.jdbc.core.JdbcTemplate#queryForMap(String)
 	 * @see org.springframework.jdbc.core.ColumnMapRowMapper
 	 */
@@ -332,7 +364,7 @@ public interface NamedParameterJdbcOperations {
 	 * @return the result Map (one entry for each column, using the column name as the key)
 	 * @throws org.springframework.dao.IncorrectResultSizeDataAccessException
 	 * if the query does not return exactly one row
-	 * @throws org.springframework.dao.DataAccessException if the query fails
+	 * @throws DataAccessException if the query fails
 	 * @see org.springframework.jdbc.core.JdbcTemplate#queryForMap(String)
 	 * @see org.springframework.jdbc.core.ColumnMapRowMapper
 	 */
@@ -348,7 +380,7 @@ public interface NamedParameterJdbcOperations {
 	 * @param elementType the required type of element in the result list
 	 * (for example, {@code Integer.class})
 	 * @return a List of objects that match the specified element type
-	 * @throws org.springframework.dao.DataAccessException if the query fails
+	 * @throws DataAccessException if the query fails
 	 * @see org.springframework.jdbc.core.JdbcTemplate#queryForList(String, Class)
 	 * @see org.springframework.jdbc.core.SingleColumnRowMapper
 	 */
@@ -366,7 +398,7 @@ public interface NamedParameterJdbcOperations {
 	 * @param elementType the required type of element in the result list
 	 * (for example, {@code Integer.class})
 	 * @return a List of objects that match the specified element type
-	 * @throws org.springframework.dao.DataAccessException if the query fails
+	 * @throws DataAccessException if the query fails
 	 * @see org.springframework.jdbc.core.JdbcTemplate#queryForList(String, Class)
 	 * @see org.springframework.jdbc.core.SingleColumnRowMapper
 	 */
@@ -383,7 +415,7 @@ public interface NamedParameterJdbcOperations {
 	 * @param sql the SQL query to execute
 	 * @param paramSource container of arguments to bind to the query
 	 * @return a List that contains a Map per row
-	 * @throws org.springframework.dao.DataAccessException if the query fails
+	 * @throws DataAccessException if the query fails
 	 * @see org.springframework.jdbc.core.JdbcTemplate#queryForList(String)
 	 */
 	List<Map<String, Object>> queryForList(String sql, SqlParameterSource paramSource) throws DataAccessException;
@@ -399,7 +431,7 @@ public interface NamedParameterJdbcOperations {
 	 * @param paramMap map of parameters to bind to the query
 	 * (leaving it to the PreparedStatement to guess the corresponding SQL type)
 	 * @return a List that contains a Map per row
-	 * @throws org.springframework.dao.DataAccessException if the query fails
+	 * @throws DataAccessException if the query fails
 	 * @see org.springframework.jdbc.core.JdbcTemplate#queryForList(String)
 	 */
 	List<Map<String, Object>> queryForList(String sql, Map<String, ?> paramMap) throws DataAccessException;
@@ -417,7 +449,7 @@ public interface NamedParameterJdbcOperations {
 	 * @param paramSource container of arguments to bind to the query
 	 * @return an SqlRowSet representation (possibly a wrapper around a
 	 * {@code javax.sql.rowset.CachedRowSet})
-	 * @throws org.springframework.dao.DataAccessException if there is any problem executing the query
+	 * @throws DataAccessException if there is any problem executing the query
 	 * @see org.springframework.jdbc.core.JdbcTemplate#queryForRowSet(String)
 	 * @see org.springframework.jdbc.core.SqlRowSetResultSetExtractor
 	 * @see javax.sql.rowset.CachedRowSet
@@ -438,7 +470,7 @@ public interface NamedParameterJdbcOperations {
 	 * (leaving it to the PreparedStatement to guess the corresponding SQL type)
 	 * @return an SqlRowSet representation (possibly a wrapper around a
 	 * {@code javax.sql.rowset.CachedRowSet})
-	 * @throws org.springframework.dao.DataAccessException if there is any problem executing the query
+	 * @throws DataAccessException if there is any problem executing the query
 	 * @see org.springframework.jdbc.core.JdbcTemplate#queryForRowSet(String)
 	 * @see org.springframework.jdbc.core.SqlRowSetResultSetExtractor
 	 * @see javax.sql.rowset.CachedRowSet
@@ -450,7 +482,7 @@ public interface NamedParameterJdbcOperations {
 	 * @param sql the SQL containing named parameters
 	 * @param paramSource container of arguments and SQL types to bind to the query
 	 * @return the number of rows affected
-	 * @throws org.springframework.dao.DataAccessException if there is any problem issuing the update
+	 * @throws DataAccessException if there is any problem issuing the update
 	 */
 	int update(String sql, SqlParameterSource paramSource) throws DataAccessException;
 
@@ -460,7 +492,7 @@ public interface NamedParameterJdbcOperations {
 	 * @param paramMap map of parameters to bind to the query
 	 * (leaving it to the PreparedStatement to guess the corresponding SQL type)
 	 * @return the number of rows affected
-	 * @throws org.springframework.dao.DataAccessException if there is any problem issuing the update
+	 * @throws DataAccessException if there is any problem issuing the update
 	 */
 	int update(String sql, Map<String, ?> paramMap) throws DataAccessException;
 
@@ -471,7 +503,7 @@ public interface NamedParameterJdbcOperations {
 	 * @param paramSource container of arguments and SQL types to bind to the query
 	 * @param generatedKeyHolder a {@link KeyHolder} that will hold the generated keys
 	 * @return the number of rows affected
-	 * @throws org.springframework.dao.DataAccessException if there is any problem issuing the update
+	 * @throws DataAccessException if there is any problem issuing the update
 	 * @see MapSqlParameterSource
 	 * @see org.springframework.jdbc.support.GeneratedKeyHolder
 	 */
@@ -486,7 +518,7 @@ public interface NamedParameterJdbcOperations {
 	 * @param generatedKeyHolder a {@link KeyHolder} that will hold the generated keys
 	 * @param keyColumnNames names of the columns that will have keys generated for them
 	 * @return the number of rows affected
-	 * @throws org.springframework.dao.DataAccessException if there is any problem issuing the update
+	 * @throws DataAccessException if there is any problem issuing the update
 	 * @see MapSqlParameterSource
 	 * @see org.springframework.jdbc.support.GeneratedKeyHolder
 	 */
@@ -498,6 +530,7 @@ public interface NamedParameterJdbcOperations {
 	 * @param sql the SQL statement to execute
 	 * @param batchValues the array of Maps containing the batch of arguments for the query
 	 * @return an array containing the numbers of rows affected by each update in the batch
+	 * @throws DataAccessException if there is any problem issuing the update
 	 */
 	int[] batchUpdate(String sql, Map<String, ?>[] batchValues);
 
@@ -506,6 +539,7 @@ public interface NamedParameterJdbcOperations {
 	 * @param sql the SQL statement to execute
 	 * @param batchArgs the array of {@link SqlParameterSource} containing the batch of arguments for the query
 	 * @return an array containing the numbers of rows affected by each update in the batch
+	 * @throws DataAccessException if there is any problem issuing the update
 	 */
 	int[] batchUpdate(String sql, SqlParameterSource[] batchArgs);
 

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

@@ -22,6 +22,7 @@ import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.function.Consumer;
+import java.util.stream.Stream;
 
 import javax.sql.DataSource;
 
@@ -228,6 +229,20 @@ public class NamedParameterJdbcTemplate implements NamedParameterJdbcOperations
 		return query(sql, EmptySqlParameterSource.INSTANCE, rowMapper);
 	}
 
+	@Override
+	public <T> Stream<T> queryForStream(String sql, SqlParameterSource paramSource, RowMapper<T> rowMapper)
+			throws DataAccessException {
+
+		return getJdbcOperations().queryForStream(getPreparedStatementCreator(sql, paramSource), rowMapper);
+	}
+
+	@Override
+	public <T> Stream<T> queryForStream(String sql, Map<String, ?> paramMap, RowMapper<T> rowMapper)
+			throws DataAccessException {
+
+		return queryForStream(sql, new MapSqlParameterSource(paramMap), rowMapper);
+	}
+
 	@Override
 	@Nullable
 	public <T> T queryForObject(String sql, SqlParameterSource paramSource, RowMapper<T> rowMapper)

spring-jdbc/src/test/java/org/springframework/jdbc/core/JdbcTemplateQueryTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -26,6 +26,8 @@ import java.sql.SQLException;
 import java.sql.Statement;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Stream;
 
 import javax.sql.DataSource;
 
@@ -164,8 +166,24 @@ public class JdbcTemplateQueryTests {
 				return rs.getInt(1);
 			}
 		});
-		boolean condition = o instanceof Integer;
+		assertThat(o instanceof Integer).as("Correct result type").isTrue();
-		assertThat(condition).as("Correct result type").isTrue();
+		verify(this.resultSet).close();
+		verify(this.statement).close();
+	}
+
+	@Test
+	public void testQueryForStreamWithRowMapper() throws Exception {
+		String sql = "SELECT AGE FROM CUSTMR WHERE ID = 3";
+		given(this.resultSet.next()).willReturn(true, false);
+		given(this.resultSet.getInt(1)).willReturn(22);
+		AtomicInteger count = new AtomicInteger();
+		try (Stream<Integer> s = this.template.queryForStream(sql, (rs, rowNum) -> rs.getInt(1))) {
+			s.forEach(val -> {
+				count.incrementAndGet();
+				assertThat(val).isEqualTo(22);
+			});
+		}
+		assertThat(count.get()).isEqualTo(1);
 		verify(this.resultSet).close();
 		verify(this.statement).close();
 	}
@@ -278,7 +296,7 @@ public class JdbcTemplateQueryTests {
 	private void doTestQueryForListWithArgs(String sql) throws Exception {
 		given(this.resultSet.next()).willReturn(true, true, false);
 		given(this.resultSet.getObject(1)).willReturn(11, 12);
-		List<Map<String, Object>> li = this.template.queryForList(sql, new Object[] {3});
+		List<Map<String, Object>> li = this.template.queryForList(sql, 3);
 		assertThat(li.size()).as("All rows returned").isEqualTo(2);
 		assertThat(((Integer) li.get(0).get("age")).intValue()).as("First row is Integer").isEqualTo(11);
 		assertThat(((Integer) li.get(1).get("age")).intValue()).as("Second row is Integer").isEqualTo(12);
@@ -291,7 +309,7 @@ public class JdbcTemplateQueryTests {
 	public void testQueryForListWithArgsAndEmptyResult() throws Exception {
 		String sql = "SELECT AGE FROM CUSTMR WHERE ID < ?";
 		given(this.resultSet.next()).willReturn(false);
-		List<Map<String, Object>> li = this.template.queryForList(sql, new Object[] {3});
+		List<Map<String, Object>> li = this.template.queryForList(sql, 3);
 		assertThat(li.size()).as("All rows returned").isEqualTo(0);
 		verify(this.preparedStatement).setObject(1, 3);
 		verify(this.resultSet).close();
@@ -303,7 +321,7 @@ public class JdbcTemplateQueryTests {
 		String sql = "SELECT AGE FROM CUSTMR WHERE ID < ?";
 		given(this.resultSet.next()).willReturn(true, false);
 		given(this.resultSet.getObject(1)).willReturn(11);
-		List<Map<String, Object>> li = this.template.queryForList(sql, new Object[] {3});
+		List<Map<String, Object>> li = this.template.queryForList(sql, 3);
 		assertThat(li.size()).as("All rows returned").isEqualTo(1);
 		assertThat(((Integer) li.get(0).get("age")).intValue()).as("First row is Integer").isEqualTo(11);
 		verify(this.preparedStatement).setObject(1, 3);
@@ -316,7 +334,7 @@ public class JdbcTemplateQueryTests {
 		String sql = "SELECT AGE FROM CUSTMR WHERE ID < ?";
 		given(this.resultSet.next()).willReturn(true, false);
 		given(this.resultSet.getInt(1)).willReturn(11);
-		List<Integer> li = this.template.queryForList(sql, new Object[] {3}, Integer.class);
+		List<Integer> li = this.template.queryForList(sql, Integer.class, 3);
 		assertThat(li.size()).as("All rows returned").isEqualTo(1);
 		assertThat(li.get(0).intValue()).as("First row is Integer").isEqualTo(11);
 		verify(this.preparedStatement).setObject(1, 3);
@@ -329,7 +347,7 @@ public class JdbcTemplateQueryTests {
 		String sql = "SELECT AGE FROM CUSTMR WHERE ID < ?";
 		given(this.resultSet.next()).willReturn(true, false);
 		given(this.resultSet.getObject(1)).willReturn(11);
-		Map<String, Object> map = this.template.queryForMap(sql, new Object[] {3});
+		Map<String, Object> map = this.template.queryForMap(sql, 3);
 		assertThat(((Integer) map.get("age")).intValue()).as("Row is Integer").isEqualTo(11);
 		verify(this.preparedStatement).setObject(1, 3);
 		verify(this.resultSet).close();
@@ -341,14 +359,26 @@ public class JdbcTemplateQueryTests {
 		String sql = "SELECT AGE FROM CUSTMR WHERE ID = ?";
 		given(this.resultSet.next()).willReturn(true, false);
 		given(this.resultSet.getInt(1)).willReturn(22);
-		Object o = this.template.queryForObject(sql, new Object[] {3}, new RowMapper<Integer>() {
+		Object o = this.template.queryForObject(sql, (rs, rowNum) -> rs.getInt(1), 3);
-			@Override
+		assertThat(o instanceof Integer).as("Correct result type").isTrue();
-			public Integer mapRow(ResultSet rs, int rowNum) throws SQLException {
+		verify(this.preparedStatement).setObject(1, 3);
-				return rs.getInt(1);
+		verify(this.resultSet).close();
-			}
+		verify(this.preparedStatement).close();
-		});
+	}
-		boolean condition = o instanceof Integer;
+
-		assertThat(condition).as("Correct result type").isTrue();
+	@Test
+	public void testQueryForStreamWithArgsAndRowMapper() throws Exception {
+		String sql = "SELECT AGE FROM CUSTMR WHERE ID = ?";
+		given(this.resultSet.next()).willReturn(true, false);
+		given(this.resultSet.getInt(1)).willReturn(22);
+		AtomicInteger count = new AtomicInteger();
+		try (Stream<Integer> s = this.template.queryForStream(sql, (rs, rowNum) -> rs.getInt(1), 3)) {
+			s.forEach(val -> {
+				count.incrementAndGet();
+				assertThat(val).isEqualTo(22);
+			});
+		}
+		assertThat(count.get()).isEqualTo(1);
 		verify(this.preparedStatement).setObject(1, 3);
 		verify(this.resultSet).close();
 		verify(this.preparedStatement).close();
@@ -359,9 +389,8 @@ public class JdbcTemplateQueryTests {
 		String sql = "SELECT AGE FROM CUSTMR WHERE ID = ?";
 		given(this.resultSet.next()).willReturn(true, false);
 		given(this.resultSet.getInt(1)).willReturn(22);
-		Object o = this.template.queryForObject(sql, new Object[] {3}, Integer.class);
+		Object o = this.template.queryForObject(sql, Integer.class, 3);
-		boolean condition = o instanceof Integer;
+		assertThat(o instanceof Integer).as("Correct result type").isTrue();
-		assertThat(condition).as("Correct result type").isTrue();
 		verify(this.preparedStatement).setObject(1, 3);
 		verify(this.resultSet).close();
 		verify(this.preparedStatement).close();
@@ -372,7 +401,7 @@ public class JdbcTemplateQueryTests {
 		String sql = "SELECT AGE FROM CUSTMR WHERE ID = ?";
 		given(this.resultSet.next()).willReturn(true, false);
 		given(this.resultSet.getInt(1)).willReturn(22);
-		int i = this.template.queryForObject(sql, new Object[] {3}, Integer.class).intValue();
+		int i = this.template.queryForObject(sql, Integer.class, 3).intValue();
 		assertThat(i).as("Return of an int").isEqualTo(22);
 		verify(this.preparedStatement).setObject(1, 3);
 		verify(this.resultSet).close();
@@ -384,7 +413,7 @@ public class JdbcTemplateQueryTests {
 		String sql = "SELECT AGE FROM CUSTMR WHERE ID = ?";
 		given(this.resultSet.next()).willReturn(true, false);
 		given(this.resultSet.getLong(1)).willReturn(87L);
-		long l = this.template.queryForObject(sql, new Object[] {3}, Long.class).longValue();
+		long l = this.template.queryForObject(sql, Long.class, 3).longValue();
 		assertThat(l).as("Return of a long").isEqualTo(87);
 		verify(this.preparedStatement).setObject(1, 3);
 		verify(this.resultSet).close();

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -28,6 +28,8 @@ import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Stream;
 
 import javax.sql.DataSource;
 
@@ -237,6 +239,7 @@ public class NamedParameterJdbcTemplateTests {
 		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();
 	}
@@ -259,6 +262,7 @@ public class NamedParameterJdbcTemplateTests {
 		assertThat(cust.getId() == 1).as("Customer id was assigned correctly").isTrue();
 		assertThat(cust.getForename().equals("rod")).as("Customer forename was assigned correctly").isTrue();
 		verify(connection).prepareStatement(SELECT_NO_PARAMETERS);
+		verify(resultSet).close();
 		verify(preparedStatement).close();
 		verify(connection).close();
 	}
@@ -285,6 +289,7 @@ public class NamedParameterJdbcTemplateTests {
 		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();
 	}
@@ -307,6 +312,7 @@ public class NamedParameterJdbcTemplateTests {
 		assertThat(customers.get(0).getId() == 1).as("Customer id was assigned correctly").isTrue();
 		assertThat(customers.get(0).getForename().equals("rod")).as("Customer forename was assigned correctly").isTrue();
 		verify(connection).prepareStatement(SELECT_NO_PARAMETERS);
+		verify(resultSet).close();
 		verify(preparedStatement).close();
 		verify(connection).close();
 	}
@@ -326,12 +332,14 @@ public class NamedParameterJdbcTemplateTests {
 					cust.setForename(rs.getString(COLUMN_NAMES[1]));
 					return cust;
 				});
+
 		assertThat(customers.size()).isEqualTo(1);
 		assertThat(customers.get(0).getId() == 1).as("Customer id was assigned correctly").isTrue();
 		assertThat(customers.get(0).getForename().equals("rod")).as("Customer forename was assigned correctly").isTrue();
 		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();
 	}
@@ -349,10 +357,12 @@ public class NamedParameterJdbcTemplateTests {
 					cust.setForename(rs.getString(COLUMN_NAMES[1]));
 					return cust;
 				});
+
 		assertThat(customers.size()).isEqualTo(1);
 		assertThat(customers.get(0).getId() == 1).as("Customer id was assigned correctly").isTrue();
 		assertThat(customers.get(0).getForename().equals("rod")).as("Customer forename was assigned correctly").isTrue();
 		verify(connection).prepareStatement(SELECT_NO_PARAMETERS);
+		verify(resultSet).close();
 		verify(preparedStatement).close();
 		verify(connection).close();
 	}
@@ -365,6 +375,7 @@ public class NamedParameterJdbcTemplateTests {
 
 		params.put("id", new SqlParameterValue(Types.DECIMAL, 1));
 		params.put("country", "UK");
+
 		Customer cust = namedParameterTemplate.queryForObject(SELECT_NAMED_PARAMETERS, params,
 				(rs, rownum) -> {
 					Customer cust1 = new Customer();
@@ -372,11 +383,46 @@ public class NamedParameterJdbcTemplateTests {
 					cust1.setForename(rs.getString(COLUMN_NAMES[1]));
 					return cust1;
 				});
+
 		assertThat(cust.getId() == 1).as("Customer id was assigned correctly").isTrue();
 		assertThat(cust.getForename().equals("rod")).as("Customer forename was assigned correctly").isTrue();
 		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 = namedParameterTemplate.queryForStream(SELECT_NAMED_PARAMETERS, params,
+				(rs, rownum) -> {
+					Customer cust1 = new Customer();
+					cust1.setId(rs.getInt(COLUMN_NAMES[0]));
+					cust1.setForename(rs.getString(COLUMN_NAMES[1]));
+					return cust1;
+				})) {
+			s.forEach(cust -> {
+				count.incrementAndGet();
+				assertThat(cust.getId() == 1).as("Customer id was assigned correctly").isTrue();
+				assertThat(cust.getForename().equals("rod")).as("Customer forename was assigned correctly").isTrue();
+			});
+		}
+
+		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();
 	}