Add support for streamed query results.

Use queryForStream for streamed query results. ResultSetExtractor is ignored because it cannot be used together with streams.

Original pull request #176
Closes #356

Author: denniseffing

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

@@ -18,6 +18,7 @@
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.List;
+import java.util.stream.Stream;
 
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.dao.EmptyResultDataAccessException;
@@ -40,6 +41,7 @@
  * @author Oliver Gierke
  * @author Maciej Walkowiak
  * @author Mark Paluch
+ * @author Dennis Effing
  * @since 2.0
  */
 public abstract class AbstractJdbcQuery implements RepositoryQuery {
@@ -88,10 +90,14 @@ protected JdbcQueryExecution<?> getQueryExecution(JdbcQueryMethod queryMethod,
 			return createModifyingQueryExecutor();
 		}
 
-		if (queryMethod.isCollectionQuery() || queryMethod.isStreamQuery()) {
+		if (queryMethod.isCollectionQuery()) {
 			return extractor != null ? getQueryExecution(extractor) : collectionQuery(rowMapper);
 		}
 
+		if (queryMethod.isStreamQuery()) {
+			return streamQuery(rowMapper);
+		}
+
 		return extractor != null ? getQueryExecution(extractor) : singleObjectQuery(rowMapper);
 	}
 
@@ -140,6 +146,10 @@ protected Class<?> resolveTypeToRead(ResultProcessor resultProcessor) {
 				: returnedType.getReturnedType();
 	}
 
+	private <T> JdbcQueryExecution<Stream<T>> streamQuery(RowMapper<T> rowMapper) {
+		return (query, parameters) -> operations.queryForStream(query, parameters, rowMapper);
+	}
+
 	private <T> JdbcQueryExecution<T> getQueryExecution(ResultSetExtractor<T> resultSetExtractor) {
 		return (query, parameters) -> operations.query(query, parameters, resultSetExtractor);
 	}

Diff for: spring-data-jdbc/src/test/java/org/springframework/data/jdbc/repository/query/QueryAnnotationHsqlIntegrationTests.java

@@ -51,6 +51,7 @@
  * @author Jens Schauder
  * @author Kazuki Shimizu
  * @author Mark Paluch
+ * @author Dennis Effing
  */
 @Transactional
 @ActiveProfiles("hsql")
@@ -173,6 +174,21 @@ public void executeCustomQueryWithReturnTypeIsStream() {
 				.containsExactlyInAnyOrder("a", "b");
 	}
 
+	@Test // DATAJDBC-356
+	public void executeCustomQueryWithNamedParameterAndReturnTypeIsStream() {
+
+		repository.save(dummyEntity("a"));
+		repository.save(dummyEntity("b"));
+		repository.save(dummyEntity("c"));
+
+		Stream<DummyEntity> entities = repository.findByNamedRangeWithNamedParameterAndReturnTypeIsStream("a", "c");
+
+		assertThat(entities) //
+				.extracting(e -> e.name) //
+				.containsExactlyInAnyOrder("b");
+
+	}
+
 	@Test // DATAJDBC-175
 	public void executeCustomQueryWithReturnTypeIsNumber() {
 
@@ -292,6 +308,10 @@ private interface DummyEntityRepository extends CrudRepository<DummyEntity, Long
 		@Query("SELECT * FROM DUMMY_ENTITY")
 		Stream<DummyEntity> findAllWithReturnTypeIsStream();
 
+		@Query("SELECT * FROM DUMMY_ENTITY WHERE name  < :upper and name > :lower")
+		Stream<DummyEntity> findByNamedRangeWithNamedParameterAndReturnTypeIsStream(@Param("lower") String lower,
+				@Param("upper") String upper);
+
 		// DATAJDBC-175
 		@Query("SELECT count(*) FROM DUMMY_ENTITY WHERE name like concat('%', :name, '%')")
 		int countByNameContaining(@Param("name") String name);

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

@@ -42,6 +42,7 @@
 import org.springframework.jdbc.core.ResultSetExtractor;
 import org.springframework.jdbc.core.RowMapper;
 import org.springframework.jdbc.core.namedparam.NamedParameterJdbcOperations;
+import org.springframework.jdbc.core.namedparam.SqlParameterSource;
 import org.springframework.util.ReflectionUtils;
 
 /**
@@ -52,6 +53,7 @@
  * @author Maciej Walkowiak
  * @author Evgeni Dimitrov
  * @author Mark Paluch
+ * @author Dennis Effing
  */
 public class StringBasedJdbcQueryUnitTests {
 
@@ -127,6 +129,17 @@ public void customResultSetExtractorAndRowMapperGetCombined() {
 						"RowMapper is not expected to be custom");
 	}
 
+	@Test // DATAJDBC-356
+	void streamQueryCallsQueryForStreamOnOperations() {
+		doReturn(true).when(queryMethod).isStreamQuery();
+		doReturn("some sql statement").when(queryMethod).getDeclaredQuery();
+
+		StringBasedJdbcQuery subject = createQuery();
+		subject.execute(new Object[] {});
+
+		verify(operations).queryForStream(eq("some sql statement"), any(SqlParameterSource.class), eq(defaultRowMapper));
+	}
+
 	@Test // GH-774
 	public void sliceQueryNotSupported() {
 
@@ -183,6 +196,12 @@ interface MyRepository extends Repository<Object, Long> {
 
 	}
 
+	/**
+	 * The whole purpose of this method is to easily generate a {@link DefaultParameters} instance during test setup.
+	 */
+	@SuppressWarnings("unused")
+	private void dummyMethod() {}
+
 	private static class CustomRowMapper implements RowMapper<Object> {
 
 		@Override