Add support for streamed query results.

Use queryForStream for streamed query results.
Since ResultSetExtractor cannot be reasonably be used together with streams it falls back to the existing collection behaviour.

Closes #578

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 extractor != null ? getQueryExecution(extractor) : 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

@@ -22,11 +22,12 @@
 import java.sql.ResultSet;
 import java.util.List;
 import java.util.Properties;
+import java.util.stream.Stream;
 
 import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-
+import org.mockito.ArgumentCaptor;
 import org.springframework.dao.DataAccessException;
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.Pageable;
@@ -39,9 +40,11 @@
 import org.springframework.data.repository.Repository;
 import org.springframework.data.repository.core.support.DefaultRepositoryMetadata;
 import org.springframework.data.repository.core.support.PropertiesBasedNamedQueries;
+import org.springframework.data.repository.query.DefaultParameters;
 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 +55,7 @@
  * @author Maciej Walkowiak
  * @author Evgeni Dimitrov
  * @author Mark Paluch
+ * @author Dennis Effing
  */
 public class StringBasedJdbcQueryUnitTests {
 
@@ -127,6 +131,28 @@ public void customResultSetExtractorAndRowMapperGetCombined() {
 						"RowMapper is not expected to be custom");
 	}
 
+	@Test // DATAJDBC-356
+	public void streamQueryCallsQueryForStreamOnOperations() {
+		JdbcQueryMethod queryMethod = createMethod("findAllWithStreamReturnType");
+		StringBasedJdbcQuery query = createQuery(queryMethod);
+
+		query.execute(new Object[] {});
+
+		verify(operations).queryForStream(eq("some sql statement"), any(SqlParameterSource.class), any(RowMapper.class));
+	}
+
+	@Test // DATAJDBC-356
+	void streamQueryFallsBackToCollectionQueryWhenCustomResultSetExtractorIsSpecified() {
+		JdbcQueryMethod queryMethod = createMethod("findAllWithStreamReturnTypeAndResultSetExtractor");
+		StringBasedJdbcQuery query = createQuery(queryMethod);
+
+		query.execute(new Object[] {});
+
+		ArgumentCaptor<ResultSetExtractor> captor = ArgumentCaptor.forClass(ResultSetExtractor.class);
+		verify(operations).query(eq("some sql statement"), any(SqlParameterSource.class), captor.capture());
+		assertThat(captor.getValue()).isInstanceOf(CustomResultSetExtractor.class);
+	}
+
 	@Test // GH-774
 	public void sliceQueryNotSupported() {
 
@@ -173,6 +199,12 @@ interface MyRepository extends Repository<Object, Long> {
 				resultSetExtractorClass = CustomResultSetExtractor.class)
 		List<Object> findAllWithCustomRowMapperAndResultSetExtractor();
 
+		@Query(value = "some sql statement")
+		Stream<Object> findAllWithStreamReturnType();
+
+		@Query(value = "some sql statement", resultSetExtractorClass = CustomResultSetExtractor.class)
+		Stream<Object> findAllWithStreamReturnTypeAndResultSetExtractor();
+
 		List<Object> noAnnotation();
 
 		@Query(value = "some sql statement")