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 spring-projects#176
Closes spring-projects#356

Author: romansey

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

@@ -16,6 +16,7 @@
 package org.springframework.data.jdbc.repository.query;
 
 import java.util.List;
+import java.util.stream.Stream;
 
 import org.springframework.dao.EmptyResultDataAccessException;
 import org.springframework.data.repository.query.RepositoryQuery;
@@ -89,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);
 	}
 
@@ -123,6 +128,10 @@ private <T> JdbcQueryExecution<List<T>> collectionQuery(RowMapper<T> rowMapper)
 		return getQueryExecution(new RowMapperResultSetExtractor<>(rowMapper));
 	}
 
+	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

@@ -170,6 +170,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() {
 
@@ -289,6 +304,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

@@ -35,6 +35,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;
 
 /**
  * Unit tests for {@link StringBasedJdbcQuery}.
@@ -151,6 +152,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));
+	}
+
 	/**
 	 * The whole purpose of this method is to easily generate a {@link DefaultParameters} instance during test setup.
 	 */