From 5a3d92053293c8be4a405aa33ad8b413671a6f3a Mon Sep 17 00:00:00 2001 From: Thomas Darimont Date: Tue, 24 Feb 2015 23:01:53 +0100 Subject: [PATCH] DATAJPA-677 - Add support for Java 8 Stream in repository finder methods. Added support for Streaming query results in Hibernate, EclipseLink and OpenJPA by falling back to store specific APIs to execute the query. Original pull request: #136. --- .../jpa/provider/PersistenceProvider.java | 222 +++++++++++++++++- .../repository/query/AbstractJpaQuery.java | 5 +- .../repository/query/JpaQueryExecution.java | 24 +- .../jpa/repository/UserRepositoryTests.java | 66 +++++- .../jpa/repository/sample/UserRepository.java | 12 + 5 files changed, 324 insertions(+), 5 deletions(-) diff --git a/src/main/java/org/springframework/data/jpa/provider/PersistenceProvider.java b/src/main/java/org/springframework/data/jpa/provider/PersistenceProvider.java index aff2820e0..19697e63f 100644 --- a/src/main/java/org/springframework/data/jpa/provider/PersistenceProvider.java +++ b/src/main/java/org/springframework/data/jpa/provider/PersistenceProvider.java @@ -1,5 +1,5 @@ /* - * Copyright 2008-2014 the original author or authors. + * Copyright 2008-2015 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. @@ -21,17 +21,31 @@ import static org.springframework.data.jpa.provider.PersistenceProvider.Constant import java.util.Arrays; import java.util.Collection; import java.util.Collections; +import java.util.Iterator; +import java.util.List; import javax.persistence.EntityManager; import javax.persistence.Query; import javax.persistence.metamodel.Metamodel; import org.apache.openjpa.enhance.PersistenceCapable; +import org.apache.openjpa.persistence.OpenJPAPersistence; import org.apache.openjpa.persistence.OpenJPAQuery; +import org.apache.openjpa.persistence.jdbc.FetchDirection; +import org.apache.openjpa.persistence.jdbc.JDBCFetchPlan; +import org.apache.openjpa.persistence.jdbc.LRSSizeAlgorithm; +import org.apache.openjpa.persistence.jdbc.ResultSetType; import org.eclipse.persistence.jpa.JpaQuery; +import org.eclipse.persistence.queries.ScrollableCursor; +import org.hibernate.ScrollMode; +import org.hibernate.ScrollableResults; import org.hibernate.ejb.HibernateQuery; import org.hibernate.proxy.HibernateProxy; +import org.springframework.beans.DirectFieldAccessor; +import org.springframework.data.util.CloseableIterator; +import org.springframework.transaction.support.TransactionSynchronizationManager; import org.springframework.util.Assert; +import org.springframework.util.ClassUtils; /** * Enumeration representing persistence providers to be used. @@ -94,6 +108,11 @@ public enum PersistenceProvider implements QueryExtractor, ProxyIdAccessor { public Collection potentiallyConvertEmptyCollection(Collection collection) { return collection == null || collection.isEmpty() ? null : collection; } + + @Override + public CloseableIterator executeQueryWithResultStream(Query jpaQuery) { + return new HibernateScrollableResultsIterator(jpaQuery); + } }, /** @@ -131,6 +150,11 @@ public enum PersistenceProvider implements QueryExtractor, ProxyIdAccessor { public Collection potentiallyConvertEmptyCollection(Collection collection) { return collection == null || collection.isEmpty() ? null : collection; } + + @Override + public CloseableIterator executeQueryWithResultStream(Query jpaQuery) { + return new EclipseLinkScrollableResultsIterator(jpaQuery); + } }, /** @@ -164,6 +188,11 @@ public enum PersistenceProvider implements QueryExtractor, ProxyIdAccessor { public Object getIdentifierFrom(Object entity) { return ((PersistenceCapable) entity).pcFetchObjectId(); } + + @Override + public CloseableIterator executeQueryWithResultStream(Query jpaQuery) { + return new OpenJpaResultStreamingIterator(jpaQuery); + } }, /** @@ -310,4 +339,195 @@ public enum PersistenceProvider implements QueryExtractor, ProxyIdAccessor { public Collection potentiallyConvertEmptyCollection(Collection collection) { return collection; } + + public CloseableIterator executeQueryWithResultStream(Query jpaQuery) { + throw new UnsupportedOperationException("Streaming results is not implement for this PersistenceProvider: " + + name()); + } + + /** + * @author Thomas Darimont + * @param + * @since 1.8 + */ + @SuppressWarnings("unchecked") + static class HibernateScrollableResultsIterator implements CloseableIterator { + + private ScrollableResults scrollableResults; + + private static final boolean IS_HIBERNATE3 = ClassUtils.isPresent("org.hibernate.ejb.QueryImpl", + HibernateScrollableResultsIterator.class.getClassLoader()); + + public HibernateScrollableResultsIterator(Query jpaQuery) { + + // see http://java.dzone.com/articles/bulk-fetching-hibernate + // we could also use a Hibernate stateless session here for constructing a query + + org.hibernate.Query qry = IS_HIBERNATE3 ? extractHibernate3QueryFrom(jpaQuery) : extractHibernate4Query(jpaQuery); + + ScrollableResults scrollableResults = qry.setReadOnly( + TransactionSynchronizationManager.isCurrentTransactionReadOnly()).scroll(ScrollMode.FORWARD_ONLY); + + this.scrollableResults = scrollableResults; + } + + private org.hibernate.Query extractHibernate4Query(Query jpaQuery) { + + Object queryImpl = jpaQuery; + if (jpaQuery.getClass().getName().equals("org.hibernate.jpa.criteria.compile.CriteriaQueryTypeQueryAdapter")) { + queryImpl = new DirectFieldAccessor(jpaQuery).getPropertyValue("jpqlQuery"); + } + + return extractHibernateQueryFromQueryImpl(queryImpl); + } + + private org.hibernate.Query extractHibernate3QueryFrom(Query jpaQuery) { + + Object queryImpl = jpaQuery; + if (jpaQuery.getClass().isAnonymousClass() + && jpaQuery.getClass().getEnclosingClass().getName() + .equals("org.hibernate.ejb.criteria.CriteriaQueryCompiler")) { + queryImpl = new DirectFieldAccessor(jpaQuery).getPropertyValue("val$jpaqlQuery"); + } + + return extractHibernateQueryFromQueryImpl(queryImpl); + } + + private org.hibernate.Query extractHibernateQueryFromQueryImpl(Object queryImpl) { + return (org.hibernate.Query) new DirectFieldAccessor(queryImpl).getPropertyValue("query"); + } + + @Override + public T next() { + + Object item = scrollableResults.get()[0]; + + return (T) item; + } + + @Override + public boolean hasNext() { + + if (scrollableResults == null) { + return false; + } + + return scrollableResults.next(); + } + + @Override + public void close() { + + if (scrollableResults == null) { + return; + } + + try { + scrollableResults.close(); + } finally { + scrollableResults = null; + } + } + } + + /** + * @author Thomas Darimont + * @param + * @since 1.8 + */ + @SuppressWarnings("unchecked") + static class EclipseLinkScrollableResultsIterator implements CloseableIterator { + + private ScrollableCursor scrollableCursor; + + public EclipseLinkScrollableResultsIterator(Query jpaQuery) { + + jpaQuery.setHint("eclipselink.cursor.scrollable", true); + this.scrollableCursor = (ScrollableCursor) jpaQuery.getSingleResult(); + } + + @Override + public boolean hasNext() { + + if (scrollableCursor == null) { + return false; + } + + return scrollableCursor.hasNext(); + } + + @Override + public T next() { + + Object item = scrollableCursor.next(); + return (T) item; + } + + @Override + public void close() { + + if (scrollableCursor == null) { + return; + } + + try { + scrollableCursor.close(); + } finally { + scrollableCursor = null; + } + } + } + + /** + * @author Thomas Darimont + * @param + * @since 1.8 + */ + static class OpenJpaResultStreamingIterator implements CloseableIterator { + + private Iterator iterator; + + @SuppressWarnings({ "rawtypes", "unchecked" }) + public OpenJpaResultStreamingIterator(Query jpaQuery) { + + OpenJPAQuery kq = OpenJPAPersistence.cast(jpaQuery); + JDBCFetchPlan fetch = (JDBCFetchPlan) kq.getFetchPlan(); + fetch.setFetchBatchSize(20); + fetch.setResultSetType(ResultSetType.SCROLL_SENSITIVE); + fetch.setFetchDirection(FetchDirection.FORWARD); + fetch.setLRSSizeAlgorithm(LRSSizeAlgorithm.LAST); + + List resultList = kq.getResultList(); + iterator = resultList.iterator(); + } + + @Override + public boolean hasNext() { + + if (iterator == null) { + return false; + } + + return iterator.hasNext(); + } + + @Override + public T next() { + return iterator.next(); + } + + @Override + public void close() { + if (iterator == null) { + return; + } + + try { + OpenJPAPersistence.close(iterator); + } finally { + iterator = null; + } + } + + } } diff --git a/src/main/java/org/springframework/data/jpa/repository/query/AbstractJpaQuery.java b/src/main/java/org/springframework/data/jpa/repository/query/AbstractJpaQuery.java index 131d22a3d..ebd806324 100644 --- a/src/main/java/org/springframework/data/jpa/repository/query/AbstractJpaQuery.java +++ b/src/main/java/org/springframework/data/jpa/repository/query/AbstractJpaQuery.java @@ -28,6 +28,7 @@ import org.springframework.data.jpa.repository.query.JpaQueryExecution.PagedExec import org.springframework.data.jpa.repository.query.JpaQueryExecution.ProcedureExecution; import org.springframework.data.jpa.repository.query.JpaQueryExecution.SingleEntityExecution; import org.springframework.data.jpa.repository.query.JpaQueryExecution.SlicedExecution; +import org.springframework.data.jpa.repository.query.JpaQueryExecution.StreamExecution; import org.springframework.data.repository.query.RepositoryQuery; import org.springframework.util.Assert; @@ -99,7 +100,9 @@ public abstract class AbstractJpaQuery implements RepositoryQuery { protected JpaQueryExecution getExecution() { - if (method.isProcedureQuery()) { + if (method.isStreamQuery()) { + return new StreamExecution(); + } else if (method.isProcedureQuery()) { return new ProcedureExecution(); } else if (method.isCollectionQuery()) { return new CollectionExecution(); diff --git a/src/main/java/org/springframework/data/jpa/repository/query/JpaQueryExecution.java b/src/main/java/org/springframework/data/jpa/repository/query/JpaQueryExecution.java index 768d6d086..6adfbdb37 100644 --- a/src/main/java/org/springframework/data/jpa/repository/query/JpaQueryExecution.java +++ b/src/main/java/org/springframework/data/jpa/repository/query/JpaQueryExecution.java @@ -1,5 +1,5 @@ /* - * Copyright 2008-2014 the original author or authors. + * Copyright 2008-2015 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. @@ -31,9 +31,12 @@ import org.springframework.data.domain.PageImpl; import org.springframework.data.domain.Pageable; import org.springframework.data.domain.Slice; import org.springframework.data.domain.SliceImpl; +import org.springframework.data.jpa.provider.PersistenceProvider; import org.springframework.data.repository.query.ParameterAccessor; import org.springframework.data.repository.query.Parameters; import org.springframework.data.repository.query.ParametersParameterAccessor; +import org.springframework.data.util.CloseableIterator; +import org.springframework.data.util.Java8StreamUtils; import org.springframework.util.Assert; /** @@ -298,4 +301,23 @@ public abstract class JpaQueryExecution { return storedProcedureJpaQuery.extractOutputValue(storedProcedure); } } + + /** + * {@link Execution} executing a Java 8 Stream. + * + * @author Thomas Darimont + * @since 1.8 + */ + static class StreamExecution extends JpaQueryExecution { + + @Override + protected Object doExecute(final AbstractJpaQuery query, Object[] values) { + + Query jpaQuery = query.createQuery(values); + PersistenceProvider persistenceProvider = PersistenceProvider.fromEntityManager(query.getEntityManager()); + CloseableIterator iter = persistenceProvider.executeQueryWithResultStream(jpaQuery); + + return Java8StreamUtils.createStreamFromIterator(iter); + } + } } diff --git a/src/test/java/org/springframework/data/jpa/repository/UserRepositoryTests.java b/src/test/java/org/springframework/data/jpa/repository/UserRepositoryTests.java index 5f2392856..76f3350ed 100644 --- a/src/test/java/org/springframework/data/jpa/repository/UserRepositoryTests.java +++ b/src/test/java/org/springframework/data/jpa/repository/UserRepositoryTests.java @@ -28,6 +28,8 @@ import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.Set; +import java.util.function.Consumer; +import java.util.stream.Stream; import javax.persistence.EntityManager; import javax.persistence.PersistenceContext; @@ -1751,8 +1753,8 @@ public class UserRepositoryTests { assertThat(users.getContent(), hasSize(2)); assertThat(users.getContent().get(0), is(thirdUser)); assertThat(users.getContent().get(1), is(fourthUser)); - } - + } + /** * @see DATAJPA-629 */ @@ -1816,6 +1818,66 @@ public class UserRepositoryTests { assertThat(users.get(0), is(secondUser)); } + /** + * @see DATAJPA-677 + */ + @Test + public void shouldSupportJava8StreamsForRepositoryFinderMethods() { + + flushTestUsers(); + + Stream stream = repository.findAllByCustomQueryAndStream(); + + final List users = new ArrayList(); + + try { + stream.forEach(new Consumer() { + + @Override + public void accept(User user) { + + // System.out.printf("%s%n", user); + users.add(user); + } + + }); + } finally { + stream.close(); + } + + assertThat(users, hasSize(4)); + } + + /** + * @see DATAJPA-677 + */ + @Test + public void shouldSupportJava8StreamsForRepositoryDerivedFinderMethods() { + + flushTestUsers(); + + Stream stream = repository.readAllByFirstnameNotNull(); + + final List users = new ArrayList(); + + try { + stream.forEach(new Consumer() { + + @Override + public void accept(User user) { + + // System.out.printf("%s%n", user); + users.add(user); + } + + }); + } finally { + stream.close(); + } + + assertThat(users, hasSize(4)); + } + private Page executeSpecWithSort(Sort sort) { flushTestUsers(); diff --git a/src/test/java/org/springframework/data/jpa/repository/sample/UserRepository.java b/src/test/java/org/springframework/data/jpa/repository/sample/UserRepository.java index 083a98718..f60c3f361 100644 --- a/src/test/java/org/springframework/data/jpa/repository/sample/UserRepository.java +++ b/src/test/java/org/springframework/data/jpa/repository/sample/UserRepository.java @@ -19,6 +19,7 @@ import java.util.Collection; import java.util.Date; import java.util.List; import java.util.Set; +import java.util.stream.Stream; import javax.persistence.EntityManager; import javax.persistence.QueryHint; @@ -557,4 +558,15 @@ public interface UserRepository extends JpaRepository, JpaSpecifi * DATAJPA-606 */ List queryByAgeInOrFirstname(Integer[] ages, String firstname); + + /** + * DATAJPA-677 + */ + @Query("select u from User u") + Stream findAllByCustomQueryAndStream(); + + /** + * DATAJPA-677 + */ + Stream readAllByFirstnameNotNull(); }