Add support for streamed query results

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

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

@@ -22,11 +22,11 @@
 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.springframework.dao.DataAccessException;
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.Pageable;
@@ -39,9 +39,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 +54,7 @@
  * @author Maciej Walkowiak
  * @author Evgeni Dimitrov
  * @author Mark Paluch
+ * @author Dennis Effing
  */
 public class StringBasedJdbcQueryUnitTests {
 
@@ -127,6 +130,16 @@ 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 // GH-774
 	public void sliceQueryNotSupported() {
 
@@ -173,6 +186,9 @@ interface MyRepository extends Repository<Object, Long> {
 				resultSetExtractorClass = CustomResultSetExtractor.class)
 		List<Object> findAllWithCustomRowMapperAndResultSetExtractor();
 
+		@Query(value = "some sql statement")
+		Stream<Object> findAllWithStreamReturnType();
+
 		List<Object> noAnnotation();
 
 		@Query(value = "some sql statement")