MappedQuery - add findEach() findEachWhile() to MappedQuery (and promote to StreamableQuery)

ebean-api/src/main/java/io/ebean/DtoQuery.java

@@ -4,8 +4,6 @@ import org.jspecify.annotations.NullMarked;
 
 import java.util.Collection;
 import java.util.List;
-import java.util.function.Consumer;
-import java.util.function.Predicate;
 
 /**
  * Query for performing native SQL queries that return DTO Bean's.
@@ -49,36 +47,6 @@ public interface DtoQuery<T> extends StreamableQuery<DtoQuery<T>, T> {
    */
   QueryIterator<T> findIterate();
 
-  /**
-   * Execute the query iterating a row at a time.
-   * <p>
-   * This streaming type query is useful for large query execution as only 1 row needs to be held in memory.
-   * </p>
-   */
-  void findEach(Consumer<T> consumer);
-
-  /**
-   * Execute the query iterating the results and batching them for the consumer.
-   * <p>
-   * This runs like findEach streaming results from the database but just collects the results
-   * into batches to pass to the consumer.
-   *
-   * @param batch    The number of dto beans to collect before given them to the consumer
-   * @param consumer The consumer to process the batch of DTO beans
-   */
-  void findEach(int batch, Consumer<List<T>> consumer);
-
-  /**
-   * Execute the query iterating a row at a time with the ability to stop consuming part way through.
-   * <p>
-   * Returning false after processing a row stops the iteration through the query results.
-   * </p>
-   * <p>
-   * This streaming type query is useful for large query execution as only 1 row needs to be held in memory.
-   * </p>
-   */
-  void findEachWhile(Predicate<T> consumer);
-
   /**
    * Bind all the parameters using index positions.
    * <p>

ebean-api/src/main/java/io/ebean/MappedQuery.java

@@ -3,6 +3,8 @@ package io.ebean;
 import org.jspecify.annotations.NullMarked;
 
 import java.util.List;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
 import java.util.stream.Stream;
 
 /**
@@ -62,4 +64,48 @@ public interface MappedQuery<D> extends StreamableQuery<MappedQuery<D>, D> {
   @Override
   Stream<D> findStream();
 
+  /**
+   * Execute the query processing the mapped DTOs one at a time.
+   * <p>
+   * Mirrors {@link QueryBuilder#findEach(Consumer)} - the underlying entity graph query is
+   * streamed one entity at a time and each entity is mapped to its target DTO lazily as it is
+   * consumed, sharing one {@link DtoMapContext} across the whole callback so that repeated
+   * references to the same source entity still de-duplicate to the same DTO instance.
+   * <p>
+   * This method is appropriate to process very large query results as the mapped DTOs are
+   * consumed one at a time and do not need to be held in memory (unlike {@link #findList()}).
+   *
+   * @param consumer the consumer used to process the mapped DTOs.
+   */
+  @Override
+  void findEach(Consumer<D> consumer);
+  /**
+   * Execute findEach streaming query batching the mapped DTOs for consuming.
+   * <p>
+   * Mirrors {@link QueryBuilder#findEach(int, Consumer)} - typically used when we want to do
+   * further processing on the mapped DTOs in batch form, for example 100 at a time. Each batch
+   * shares one {@link DtoMapContext} with the rest of the query so that repeated references to
+   * the same source entity still de-duplicate to the same DTO instance.
+   *
+   * @param batch    The number of mapped DTOs processed in the batch
+   * @param consumer Process the batch of mapped DTOs
+   */
+  @Override
+  void findEach(int batch, Consumer<List<D>> consumer);
+
+  /**
+   * Execute the query using callbacks to process the resulting mapped DTOs one at a time,
+   * with the ability to stop processing part way through.
+   * <p>
+   * Mirrors {@link QueryBuilder#findEachWhile(Predicate)} - returning {@code false} after
+   * processing a DTO stops the iteration through the query results. Sharing one
+   * {@link DtoMapContext} across the whole callback so that repeated references to the same
+   * source entity still de-duplicate to the same DTO instance.
+   *
+   * @param consumer the consumer used to process the mapped DTOs, returning {@code false} to
+   *                 stop processing.
+   */
+  @Override
+  void findEachWhile(Predicate<D> consumer);
+
 }

ebean-api/src/main/java/io/ebean/QueryBuilder.java

@@ -10,7 +10,6 @@ import java.util.Optional;
 import java.util.Set;
 import java.util.function.BooleanSupplier;
 import java.util.function.Consumer;
-import java.util.function.Predicate;
 
 /**
  * Build and execute an ORM query.
@@ -807,84 +806,6 @@ public interface QueryBuilder<SELF extends QueryBuilder<SELF, T>, T> extends Que
    */
   <A> Set<A> findSingleAttributeSet();
 
-  /**
-   * Execute the query processing the beans one at a time.
-   * <p>
-   * This method is appropriate to process very large query results as the
-   * beans are consumed one at a time and do not need to be held in memory
-   * (unlike #findList #findSet etc)
-   * <p>
-   * Note that internally Ebean can inform the JDBC driver that it is expecting larger
-   * resultSet and specifically for MySQL this hint is required to stop it's JDBC driver
-   * from buffering the entire resultSet. As such, for smaller resultSets findList() is
-   * generally preferable.
-   * <p>
-   * Compared with #findEachWhile this will always process all the beans where as
-   * #findEachWhile provides a way to stop processing the query result early before
-   * all the beans have been read.
-   * <p>
-   * This method is functionally equivalent to findIterate() but instead of using an
-   * iterator uses the Consumer interface which is better suited to use with closures.
-   *
-   * <pre>{@code
-   *
-   *  new QCustomer()
-   *     .status.equalTo(Status.NEW)
-   *     .orderBy().id.asc()
-   *     .findEach((Customer customer) -> {
-   *
-   *       // do something with customer
-   *       System.out.println("-- visit " + customer);
-   *     });
-   *
-   * }</pre>
-   *
-   * @param consumer the consumer used to process the queried beans.
-   */
-  void findEach(Consumer<T> consumer);
-
-  /**
-   * Execute findEach streaming query batching the results for consuming.
-   * <p>
-   * This query execution will stream the results and is suited to consuming
-   * large numbers of results from the database.
-   * <p>
-   * Typically, we use this batch consumer when we want to do further processing on
-   * the beans and want to do that processing in batch form, for example - 100 at
-   * a time.
-   *
-   * @param batch    The number of beans processed in the batch
-   * @param consumer Process the batch of beans
-   */
-  void findEach(int batch, Consumer<List<T>> consumer);
-
-  /**
-   * Execute the query using callbacks to a visitor to process the resulting
-   * beans one at a time.
-   * <p>
-   * This method is functionally equivalent to findIterate() but instead of using an
-   * iterator uses the Predicate interface which is better suited to use with closures.
-   *
-   * <pre>{@code
-   *
-   *  new QCustomer()
-   *     .status.equalTo(Status.NEW)
-   *     .orderBy().id.asc()
-   *     .findEachWhile((Customer customer) -> {
-   *
-   *       // do something with customer
-   *       System.out.println("-- visit " + customer);
-   *
-   *       // return true to continue processing or false to stop
-   *       return (customer.getId() < 40);
-   *     });
-   *
-   * }</pre>
-   *
-   * @param consumer the consumer used to process the queried beans.
-   */
-  void findEachWhile(Predicate<T> consumer);
-
   /**
    * Return versions of a @History entity bean.
    * <p>

ebean-api/src/main/java/io/ebean/StreamableQuery.java

@@ -2,6 +2,9 @@ package io.ebean;
 
 import org.jspecify.annotations.NullMarked;
 
+import java.util.List;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
 import java.util.stream.Stream;
 
 /**
@@ -49,4 +52,50 @@ public interface StreamableQuery<SELF extends StreamableQuery<SELF, T>, T> exten
    */
   PagedList<T> findPagedList();
 
+  /**
+   * Execute the query processing the results one at a time.
+   * <p>
+   * This method is appropriate to process very large query results as the results are
+   * consumed one at a time and do not need to be held in memory (unlike {@link #findList()}).
+   * <p>
+   * Note that internally Ebean can inform the JDBC driver that it is expecting a larger
+   * resultSet and specifically for MySQL this hint is required to stop its JDBC driver
+   * from buffering the entire resultSet. As such, for smaller resultSets findList() is
+   * generally preferable.
+   * <p>
+   * Compared with {@link #findEachWhile(Predicate)} this will always process all the results
+   * whereas findEachWhile() provides a way to stop processing the query result early before
+   * all the results have been read.
+   *
+   * @param consumer the consumer used to process the queried results.
+   */
+  void findEach(Consumer<T> consumer);
+
+  /**
+   * Execute findEach streaming query batching the results for consuming.
+   * <p>
+   * This query execution will stream the results and is suited to consuming
+   * large numbers of results from the database.
+   * <p>
+   * Typically, we use this batch consumer when we want to do further processing on
+   * the results and want to do that processing in batch form, for example - 100 at
+   * a time.
+   *
+   * @param batch    The number of results processed in the batch
+   * @param consumer Process the batch of results
+   */
+  void findEach(int batch, Consumer<List<T>> consumer);
+
+  /**
+   * Execute the query using callbacks to process the resulting results one at a time,
+   * with the ability to stop processing part way through.
+   * <p>
+   * Returning {@code false} after processing a result stops the iteration through the
+   * query results.
+   *
+   * @param consumer the consumer used to process the queried results, returning
+   *                 {@code false} to stop processing.
+   */
+  void findEachWhile(Predicate<T> consumer);
+
 }

ebean-core/src/main/java/io/ebeaninternal/server/querydefn/DefaultMappedQuery.java

@@ -11,6 +11,8 @@ import io.ebeaninternal.api.SpiQuery;
 import java.sql.Connection;
 import java.util.List;
 import java.util.Optional;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
 import java.util.stream.Stream;
 
 /**
@@ -108,6 +110,27 @@ public final class DefaultMappedQuery<T, D> implements MappedQuery<D> {
     return query.findStream().map(source -> m.map(source, context));
   }
 
+  @Override
+  public void findEach(Consumer<D> consumer) {
+    DtoMapper<T, D> m = mapper();
+    DtoMapContext context = new DtoMapContext();
+    query.findEach(source -> consumer.accept(m.map(source, context)));
+  }
+
+  @Override
+  public void findEach(int batch, Consumer<List<D>> consumer) {
+    DtoMapper<T, D> m = mapper();
+    DtoMapContext context = new DtoMapContext();
+    query.findEach(batch, sourceBatch -> consumer.accept(m.mapList(sourceBatch, context)));
+  }
+
+  @Override
+  public void findEachWhile(Predicate<D> consumer) {
+    DtoMapper<T, D> m = mapper();
+    DtoMapContext context = new DtoMapContext();
+    query.findEachWhile(source -> consumer.test(m.map(source, context)));
+  }
+
   @Override
   public MappedQuery<D> usingMaster(boolean useMaster) {
     query.usingMaster(useMaster);

tests/test-dto-mapping/src/test/java/org/tests/dtomapping/TestQueryMapTo.java

@@ -14,6 +14,7 @@ import org.tests.dtomapping.model.Contact;
 import org.tests.dtomapping.model.Customer;
 import org.tests.dtomapping.model.query.QCustomer;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
 import java.util.function.Consumer;
@@ -265,6 +266,81 @@ class TestQueryMapTo {
     assertThat(dtos.get(0).getCustomer()).isSameAs(dtos.get(1).getCustomer());
   }
 
+  @Test
+  void mapTo_findEach_expectAllDtosProcessedInOrder() {
+    Customer customerA = new Customer("EachCoA");
+    customerA.save();
+    Customer customerB = new Customer("EachCoB");
+    customerB.save();
+
+    List<String> names = new ArrayList<>();
+    DB.find(Customer.class)
+      .where().in("name", "EachCoA", "EachCoB")
+      .orderBy().asc("name")
+      .mapTo(CustomerDto.class)
+      .findEach(dto -> names.add(dto.getName()));
+
+    assertThat(names).containsExactly("EachCoA", "EachCoB");
+  }
+
+  @Test
+  void mapTo_findEach_expectIdentityDedupSharedAcrossCallback() {
+    Customer customer = new Customer("EachDedupCo");
+    customer.save();
+    new Contact("Jane", "Doe", customer).save();
+    new Contact("John", "Doe", customer).save();
+
+    // both contacts share the same underlying Customer instance - the shared DtoMapContext used
+    // across the whole findEach() call should still de-duplicate to the same nested DTO
+    List<ContactDto> dtos = new ArrayList<>();
+    DB.find(Contact.class)
+      .where().eq("customer", customer)
+      .orderBy().asc("firstName")
+      .mapTo(ContactDto.class)
+      .findEach(dtos::add);
+
+    assertThat(dtos).hasSize(2);
+    assertThat(dtos.get(0).getCustomer()).isSameAs(dtos.get(1).getCustomer());
+  }
+
+  @Test
+  void mapTo_findEachBatch_expectBatchedMappedDtos() {
+    new Customer("BatchCoA").save();
+    new Customer("BatchCoB").save();
+    new Customer("BatchCoC").save();
+
+    List<List<String>> batches = new ArrayList<>();
+    DB.find(Customer.class)
+      .where().startsWith("name", "BatchCo")
+      .orderBy().asc("name")
+      .mapTo(CustomerDto.class)
+      .findEach(2, batch -> batches.add(
+        batch.stream().map(CustomerDto::getName).collect(Collectors.toList())));
+
+    assertThat(batches).hasSize(2);
+    assertThat(batches.get(0)).containsExactly("BatchCoA", "BatchCoB");
+    assertThat(batches.get(1)).containsExactly("BatchCoC");
+  }
+
+  @Test
+  void mapTo_findEachWhile_expectStopsWhenPredicateReturnsFalse() {
+    new Customer("WhileCoA").save();
+    new Customer("WhileCoB").save();
+    new Customer("WhileCoC").save();
+
+    List<String> names = new ArrayList<>();
+    DB.find(Customer.class)
+      .where().startsWith("name", "WhileCo")
+      .orderBy().asc("name")
+      .mapTo(CustomerDto.class)
+      .findEachWhile(dto -> {
+        names.add(dto.getName());
+        return !dto.getName().equals("WhileCoB");
+      });
+
+    assertThat(names).containsExactly("WhileCoA", "WhileCoB");
+  }
+
   @Test
   void mapTo_cancel_atBegin_expectPersistenceException() {
     new Customer("CancelCo").save();