diff --git a/src/main/java/io/ebean/ExtendedServer.java b/src/main/java/io/ebean/ExtendedServer.java index ae1c1a6b6..0fc917bdf 100644 --- a/src/main/java/io/ebean/ExtendedServer.java +++ b/src/main/java/io/ebean/ExtendedServer.java @@ -13,6 +13,7 @@ import java.util.Optional; import java.util.Set; import java.util.function.Consumer; import java.util.function.Predicate; +import java.util.stream.Stream; /** * The extended API for Database. @@ -107,6 +108,24 @@ public interface ExtendedServer { @Nonnull QueryIterator findIterate(Query query, Transaction transaction); + /** + * Return the query result as a Stream using a single persistence context. + *

+ * Note that the stream needs to be closed so use with try with resources. + *

+ */ + @Nonnull + Stream findStream(Query query, Transaction transaction); + + /** + * Return the query result as a Stream (with multiple persistence contexts). + *

+ * Note that the stream needs to be closed so use with try with resources. + *

+ */ + @Nonnull + Stream findLargeStream(Query query, Transaction transaction); + /** * Execute the query visiting the each bean one at a time. *

diff --git a/src/main/java/io/ebean/Query.java b/src/main/java/io/ebean/Query.java index 0ed9f4622..a51889999 100644 --- a/src/main/java/io/ebean/Query.java +++ b/src/main/java/io/ebean/Query.java @@ -11,6 +11,7 @@ import java.util.Optional; import java.util.Set; import java.util.function.Consumer; import java.util.function.Predicate; +import java.util.stream.Stream; /** * Object relational query for finding a List, Set, Map or single entity bean. @@ -702,6 +703,49 @@ public interface Query { @Nonnull QueryIterator findIterate(); + /** + * Execute the query returning the result as a Stream. + *

+ * Note that this will hold all resulting beans in memory using a single + * persistence context. Use findLargeStream() for queries that expect to + * return a large number of results. + *

+ *
{@code
+   *
+   *  // use try with resources to ensure Stream is closed
+   *
+   *  try (Stream stream = query.findStream()) {
+   *    stream
+   *    .map(...)
+   *    .collect(...);
+   *  }
+   *
+   * }
+ */ + @Nonnull + Stream findStream(); + + /** + * Execute the query returning the result as a Stream. + *

+ * Note that this uses multiple persistence contexts such that we can use + * it with a large number of results. + *

+ *
{@code
+   *
+   *  // use try with resources to ensure Stream is closed
+   *
+   *  try (Stream stream = query.findLargeStream()) {
+   *    stream
+   *    .map(...)
+   *    .collect(...);
+   *  }
+   *
+   * }
+ */ + @Nonnull + Stream findLargeStream(); + /** * Execute the query processing the beans one at a time. *

diff --git a/src/main/java/io/ebeaninternal/server/core/DefaultServer.java b/src/main/java/io/ebeaninternal/server/core/DefaultServer.java index 04ce0a268..09d8721e8 100644 --- a/src/main/java/io/ebeaninternal/server/core/DefaultServer.java +++ b/src/main/java/io/ebeaninternal/server/core/DefaultServer.java @@ -98,11 +98,11 @@ import io.ebeaninternal.server.query.CQueryEngine; import io.ebeaninternal.server.query.CallableQueryCount; import io.ebeaninternal.server.query.CallableQueryIds; import io.ebeaninternal.server.query.CallableQueryList; +import io.ebeaninternal.server.query.DtoQueryEngine; import io.ebeaninternal.server.query.LimitOffsetPagedList; import io.ebeaninternal.server.query.QueryFutureIds; import io.ebeaninternal.server.query.QueryFutureList; import io.ebeaninternal.server.query.QueryFutureRowCount; -import io.ebeaninternal.server.query.DtoQueryEngine; import io.ebeaninternal.server.querydefn.DefaultDtoQuery; import io.ebeaninternal.server.querydefn.DefaultOrmQuery; import io.ebeaninternal.server.querydefn.DefaultOrmUpdate; @@ -136,11 +136,16 @@ import java.util.List; import java.util.Map; import java.util.Optional; import java.util.Set; +import java.util.Spliterator; import java.util.concurrent.Callable; import java.util.concurrent.ConcurrentHashMap; import java.util.function.Consumer; import java.util.function.Function; import java.util.function.Predicate; +import java.util.stream.Stream; + +import static java.util.Spliterators.spliteratorUnknownSize; +import static java.util.stream.StreamSupport.stream; /** * The default server side implementation of EbeanServer. @@ -651,7 +656,7 @@ public final class DefaultServer implements SpiServer, SpiEbeanServer { executeSql(connection, databasePlatform.truncateStatement(table)); } connection.commit(); - } catch(SQLException e) { + } catch (SQLException e) { throw new PersistenceException("Error executing truncate", e); } } @@ -1507,6 +1512,48 @@ public final class DefaultServer implements SpiServer, SpiEbeanServer { } } + @Override + public Stream findLargeStream(Query query, Transaction transaction) { + return findStreamWithSingleContext(false, query, transaction); + } + + @Override + public Stream findStream(Query query, Transaction transaction) { + return findStreamWithSingleContext(true, query, transaction); + } + + private Stream findStreamWithSingleContext(boolean singleContext, Query query, Transaction transaction) { + SpiOrmQueryRequest request = createQueryRequest(Type.ITERATE, query, transaction); + if (singleContext) { + request.setIterateSingleContext(); + } + try { + request.initTransIfRequired(); + return toStream(request.findIterate()); + } catch (RuntimeException ex) { + request.endTransIfRequired(); + throw ex; + } + } + + private Stream toStream(QueryIterator queryIterator) { + return stream(spliteratorUnknownSize(queryIterator, Spliterator.ORDERED), false) + .onClose(new QueryIteratorClose(queryIterator)); + } + + private static class QueryIteratorClose implements Runnable { + private final QueryIterator iterator; + + private QueryIteratorClose(QueryIterator iterator) { + this.iterator = iterator; + } + + @Override + public void run() { + iterator.close(); + } + } + @Override public void findEach(Query query, Consumer consumer, Transaction t) { diff --git a/src/main/java/io/ebeaninternal/server/core/OrmQueryRequest.java b/src/main/java/io/ebeaninternal/server/core/OrmQueryRequest.java index d48539a6a..fab12b976 100644 --- a/src/main/java/io/ebeaninternal/server/core/OrmQueryRequest.java +++ b/src/main/java/io/ebeaninternal/server/core/OrmQueryRequest.java @@ -87,6 +87,8 @@ public final class OrmQueryRequest extends BeanRequest implements SpiOrmQuery private Set dependentTables; + private boolean iterateSingleContext; + /** * Create the InternalQueryRequest. */ @@ -104,6 +106,15 @@ public final class OrmQueryRequest extends BeanRequest implements SpiOrmQuery return queryEngine.translate(this, bindLog, sql, e); } + @Override + public void setIterateSingleContext() { + this.iterateSingleContext = true; + } + + public boolean isIterateSingleContext() { + return iterateSingleContext; + } + @Override public boolean isDeleteByStatement() { if (!transaction.isPersistCascade() || beanDescriptor.isDeleteByStatement()) { @@ -322,11 +333,13 @@ public final class OrmQueryRequest extends BeanRequest implements SpiOrmQuery * For iterate queries reset the persistenceContext and loadContext. */ public void flushPersistenceContextOnIterate() { - persistenceContext = new DefaultPersistenceContext(); - loadContext.resetPersistenceContext(persistenceContext); - if (jsonRead != null) { - jsonRead.setPersistenceContext(persistenceContext); - jsonRead.setLoadContext(loadContext); + if (!iterateSingleContext) { + persistenceContext = new DefaultPersistenceContext(); + loadContext.resetPersistenceContext(persistenceContext); + if (jsonRead != null) { + jsonRead.setPersistenceContext(persistenceContext); + jsonRead.setLoadContext(loadContext); + } } } diff --git a/src/main/java/io/ebeaninternal/server/core/SpiOrmQueryRequest.java b/src/main/java/io/ebeaninternal/server/core/SpiOrmQueryRequest.java index 81a0f520f..807407f4b 100644 --- a/src/main/java/io/ebeaninternal/server/core/SpiOrmQueryRequest.java +++ b/src/main/java/io/ebeaninternal/server/core/SpiOrmQueryRequest.java @@ -171,4 +171,10 @@ public interface SpiOrmQueryRequest extends BeanQueryRequest, DocQueryRequ * Return true if delete by statement is allowed for this type given cascade rules etc. */ boolean isDeleteByStatement(); + + /** + * Set when we want to use a single persistence context for all beans returned + * in the query (so all beans are held in memory) + */ + void setIterateSingleContext(); } diff --git a/src/main/java/io/ebeaninternal/server/query/CQueryEngine.java b/src/main/java/io/ebeaninternal/server/query/CQueryEngine.java index 9dedbaf7a..1159356f1 100644 --- a/src/main/java/io/ebeaninternal/server/query/CQueryEngine.java +++ b/src/main/java/io/ebeaninternal/server/query/CQueryEngine.java @@ -204,12 +204,14 @@ public class CQueryEngine { CQuery cquery = queryBuilder.buildQuery(request); request.setCancelableQuery(cquery); - try { if (defaultFetchSizeFindEach > 0) { request.setDefaultFetchBuffer(defaultFetchSizeFindEach); } - if (!cquery.prepareBindExecuteQueryForwardOnly(forwardOnlyHintOnFindIterate)) { + if (request.isIterateSingleContext()) { + // expected relatively small number of results, single persistence context + cquery.prepareBindExecuteQuery(); + } else if (!cquery.prepareBindExecuteQueryForwardOnly(forwardOnlyHintOnFindIterate)) { // query has been cancelled already logger.trace("Future fetch already cancelled"); return null; diff --git a/src/main/java/io/ebeaninternal/server/query/DefaultFetchGroupQuery.java b/src/main/java/io/ebeaninternal/server/query/DefaultFetchGroupQuery.java index 8ddc0e9e6..011e9666e 100644 --- a/src/main/java/io/ebeaninternal/server/query/DefaultFetchGroupQuery.java +++ b/src/main/java/io/ebeaninternal/server/query/DefaultFetchGroupQuery.java @@ -37,6 +37,7 @@ import java.util.Optional; import java.util.Set; import java.util.function.Consumer; import java.util.function.Predicate; +import java.util.stream.Stream; /** * Implementation of FetchGroup query for use to create FetchGroup via query beans. @@ -243,6 +244,18 @@ class DefaultFetchGroupQuery implements SpiFetchGroupQuery { throw new RuntimeException("EB102: Only select() and fetch() clause is allowed on FetchGroup"); } + @Nonnull + @Override + public Stream findStream() { + throw new RuntimeException("EB102: Only select() and fetch() clause is allowed on FetchGroup"); + } + + @Nonnull + @Override + public Stream findLargeStream() { + throw new RuntimeException("EB102: Only select() and fetch() clause is allowed on FetchGroup"); + } + @Override public void findEach(Consumer consumer) { throw new RuntimeException("EB102: Only select() and fetch() clause is allowed on FetchGroup"); diff --git a/src/main/java/io/ebeaninternal/server/query/DefaultOrmQueryEngine.java b/src/main/java/io/ebeaninternal/server/query/DefaultOrmQueryEngine.java index 74c42a1d5..f82e13ebf 100644 --- a/src/main/java/io/ebeaninternal/server/query/DefaultOrmQueryEngine.java +++ b/src/main/java/io/ebeaninternal/server/query/DefaultOrmQueryEngine.java @@ -104,9 +104,7 @@ public class DefaultOrmQueryEngine implements OrmQueryEngine { @Override public QueryIterator findIterate(OrmQueryRequest request) { - // LIMITATION: You can not use QueryIterator to load bean cache - flushJdbcBatchOnQuery(request); return queryEngine.findIterate(request); } diff --git a/src/main/java/io/ebeaninternal/server/querydefn/DefaultOrmQuery.java b/src/main/java/io/ebeaninternal/server/querydefn/DefaultOrmQuery.java index 7e39dbb27..61bd4513f 100644 --- a/src/main/java/io/ebeaninternal/server/querydefn/DefaultOrmQuery.java +++ b/src/main/java/io/ebeaninternal/server/querydefn/DefaultOrmQuery.java @@ -71,6 +71,7 @@ import java.util.Optional; import java.util.Set; import java.util.function.Consumer; import java.util.function.Predicate; +import java.util.stream.Stream; /** * Default implementation of an Object Relational query. @@ -756,7 +757,7 @@ public class DefaultOrmQuery implements SpiQuery { if (underlyingList.size() == 1) { SpiExpression singleExpression = underlyingList.get(0); if (singleExpression instanceof IdInExpression) { - return new CacheIdLookup<>((IdInExpression)singleExpression); + return new CacheIdLookup<>((IdInExpression) singleExpression); } } return null; @@ -1455,7 +1456,7 @@ public class DefaultOrmQuery implements SpiQuery { @Override public Query usingTransaction(Transaction transaction) { - this.transaction = (SpiTransaction)transaction; + this.transaction = (SpiTransaction) transaction; return this; } @@ -1521,6 +1522,16 @@ public class DefaultOrmQuery implements SpiQuery { return server.findIterate(this, transaction); } + @Override + public Stream findStream() { + return server.findStream(this, transaction); + } + + @Override + public Stream findLargeStream() { + return server.findLargeStream(this, transaction); + } + @Override public List> findVersions() { this.temporalMode = TemporalMode.VERSIONS; diff --git a/src/test/java/io/ebeaninternal/api/TDSpiEbeanServer.java b/src/test/java/io/ebeaninternal/api/TDSpiEbeanServer.java index 636e4db84..6e8f93486 100644 --- a/src/test/java/io/ebeaninternal/api/TDSpiEbeanServer.java +++ b/src/test/java/io/ebeaninternal/api/TDSpiEbeanServer.java @@ -66,6 +66,7 @@ import java.util.Set; import java.util.concurrent.Callable; import java.util.function.Consumer; import java.util.function.Predicate; +import java.util.stream.Stream; /** @@ -659,6 +660,16 @@ public class TDSpiEbeanServer implements SpiEbeanServer { return null; } + @Override + public Stream findStream(Query query, Transaction transaction) { + return null; + } + + @Override + public Stream findLargeStream(Query query, Transaction transaction) { + return null; + } + @Override public void findEach(Query query, Consumer consumer, Transaction transaction) { } diff --git a/src/test/java/org/tests/query/TestQueryFindStream.java b/src/test/java/org/tests/query/TestQueryFindStream.java new file mode 100644 index 000000000..b05ada969 --- /dev/null +++ b/src/test/java/org/tests/query/TestQueryFindStream.java @@ -0,0 +1,71 @@ +package org.tests.query; + +import io.ebean.BaseTestCase; +import io.ebean.DB; +import org.junit.Test; +import org.tests.model.basic.Customer; +import org.tests.model.basic.ResetBasicData; + +import java.util.List; +import java.util.stream.Stream; + +import static java.util.stream.Collectors.toList; +import static org.assertj.core.api.Assertions.assertThat; + +public class TestQueryFindStream extends BaseTestCase { + + @Test + public void findStream_basic() { + + ResetBasicData.reset(); + + try (Stream stream = DB.find(Customer.class) + .findStream()) { + + // bad example, don't use a stream like this when we can + // use findSingleAttributeList() instead + final List namesStream = stream + .map(Customer::getName) + .collect(toList()); + + final List namesQuery = DB.find(Customer.class) + .select("name") + .findSingleAttributeList(); + + assertThat(namesStream).hasSize(namesQuery.size()); + assertThat(namesStream).containsAll(namesQuery); + } + } + + @Test + public void findLargeStream_basic() { + + ResetBasicData.reset(); + + try (Stream stream = DB.find(Customer.class) + .findLargeStream()) { + + // bad example, don't use a stream like this when we can + // use findSingleAttributeList() instead + final List namesStream = stream + .map(Customer::getName) + .collect(toList()); + + final List namesQuery = DB.find(Customer.class) + .select("name") + .findSingleAttributeList(); + + assertThat(namesStream).hasSize(namesQuery.size()); + assertThat(namesStream).containsAll(namesQuery); + } + } + + @Test + public void manualTest_findSteam_when_streamNotClosed_connectionLeak() { + + Stream stream = DB.find(Customer.class).findStream(); + // remember a steam MUST be closed or we leak resources + // comment out the close(); below to leak a connection + stream.close(); + } +} diff --git a/src/test/resources/ebean.properties b/src/test/resources/ebean.properties index 85fd153cc..c067c5ed7 100644 --- a/src/test/resources/ebean.properties +++ b/src/test/resources/ebean.properties @@ -23,6 +23,7 @@ ebean.ddl.run=true ebean.ddl.header=-- Generated by ebean ${version} at ${timestamp} ebean.packages=org.tests datasource.default=h2 +#datasource.h2.capturestacktrace=true ebean.dumpMetricsOnShutdown=true ebean.dumpMetricsOptions=sql,hash