Add support for findEach() on SqlQuery with RowMapper and scalar result

ebean-core/src/main/java/io/ebeaninternal/api/SpiEbeanServer.java

@@ -234,6 +234,11 @@ public interface SpiEbeanServer extends ExtendedServer, EbeanServer, BeanCollect
    */
   <T> List<T> findSingleAttributeList(SpiSqlQuery query, Class<T> cls);
 
+  /**
+   * SqlQuery find single attribute streaming the result to a consumer.
+   */
+  <T> void findSingleAttributeEach(SpiSqlQuery query, Class<T> cls, Consumer<T> consumer);
+
   /**
    * SqlQuery find one with mapper.
    */

ebean-core/src/main/java/io/ebeaninternal/server/core/DefaultServer.java

@@ -1606,6 +1606,11 @@ public final class DefaultServer implements SpiServer, SpiEbeanServer {
     return executeSqlQuery((req) -> req.findOneMapper(mapper), query);
   }
 
+  @Override
+  public <T> void findSingleAttributeEach(SpiSqlQuery query, Class<T> cls, Consumer<T> consumer) {
+    executeSqlQuery((req) -> req.findSingleAttributeEach(cls, consumer), query);
+  }
+
   @Override
   public <T> List<T> findSingleAttributeList(SpiSqlQuery query, Class<T> cls) {
     return executeSqlQuery((req) -> req.findSingleAttributeList(cls), query);

ebean-core/src/main/java/io/ebeaninternal/server/core/RelationalQueryEngine.java

@@ -30,12 +30,12 @@ public interface RelationalQueryEngine {
   /**
    * Find each via raw consumer.
    */
-  void findEachRow(RelationalQueryRequest request, RowConsumer mapper);
+  void findEach(RelationalQueryRequest request, RowConsumer mapper);
 
   /**
    * Find one via mapper.
    */
-  <T> T findOneMapper(RelationalQueryRequest request, RowMapper<T> mapper);
+  <T> T findOne(RelationalQueryRequest request, RowMapper<T> mapper);
 
   /**
    * Find single attribute.
@@ -47,6 +47,11 @@ public interface RelationalQueryEngine {
    */
   <T> List<T> findSingleAttributeList(RelationalQueryRequest request, Class<T> cls);
 
+  /**
+   * Find single attribute streaming the result to a consumer.
+   */
+  <T> void findSingleAttributeEach(RelationalQueryRequest request, Class<T> cls, Consumer<T> consumer);
+
   /**
    * Collect SQL query execution statistics.
    */

ebean-core/src/main/java/io/ebeaninternal/server/core/RelationalQueryRequest.java

@@ -53,7 +53,7 @@ public final class RelationalQueryRequest extends AbstractSqlQueryRequest {
 
   boolean findEachRow(RowConsumer mapper) {
     flushJdbcBatchOnQuery();
-    queryEngine.findEachRow(this, mapper);
+    queryEngine.findEach(this, mapper);
     return true;
   }
 
@@ -64,7 +64,13 @@ public final class RelationalQueryRequest extends AbstractSqlQueryRequest {
 
   <T> T findOneMapper(RowMapper<T> mapper) {
     flushJdbcBatchOnQuery();
-    return queryEngine.findOneMapper(this, mapper);
+    return queryEngine.findOne(this, mapper);
+  }
+
+  public <T> boolean findSingleAttributeEach(Class<T> cls, Consumer<T> consumer) {
+    flushJdbcBatchOnQuery();
+    queryEngine.findSingleAttributeEach(this, cls, consumer);
+    return true;
   }
 
   public <T> List<T> findSingleAttributeList(Class<T> cls) {
@@ -79,7 +85,7 @@ public final class RelationalQueryRequest extends AbstractSqlQueryRequest {
 
   public void findEach(Consumer<SqlRow> consumer) {
     flushJdbcBatchOnQuery();
-    queryEngine.findEachRow(this, (resultSet, rowNum) -> consumer.accept(createNewRow()));
+    queryEngine.findEach(this, (resultSet, rowNum) -> consumer.accept(createNewRow()));
   }
 
   public void findEachWhile(Predicate<SqlRow> consumer) {

ebean-core/src/main/java/io/ebeaninternal/server/query/DefaultRelationalQueryEngine.java

@@ -17,6 +17,7 @@ import io.ebeaninternal.server.persist.Binder;
 import javax.persistence.PersistenceException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.function.Consumer;
 import java.util.function.Predicate;
 
 /**
@@ -59,7 +60,7 @@ public class DefaultRelationalQueryEngine implements RelationalQueryEngine {
   }
 
   @Override
-  public void findEachRow(RelationalQueryRequest request, RowConsumer consumer) {
+  public void findEach(RelationalQueryRequest request, RowConsumer consumer) {
     try {
       request.executeSql(binder, SpiQuery.Type.ITERATE);
       request.mapEach(consumer);
@@ -93,7 +94,7 @@ public class DefaultRelationalQueryEngine implements RelationalQueryEngine {
   }
 
   @Override
-  public <T> T findOneMapper(RelationalQueryRequest request, RowMapper<T> mapper) {
+  public <T> T findOne(RelationalQueryRequest request, RowMapper<T> mapper) {
     try {
       request.executeSql(binder, SpiQuery.Type.BEAN);
       T value = request.mapOne(mapper);
@@ -170,4 +171,23 @@ public class DefaultRelationalQueryEngine implements RelationalQueryEngine {
     }
   }
 
+  @SuppressWarnings("unchecked")
+  @Override
+  public <T> void findSingleAttributeEach(RelationalQueryRequest request, Class<T> cls, Consumer<T> consumer) {
+    ScalarType<T> scalarType = (ScalarType<T>) binder.getScalarType(cls);
+    try {
+      request.executeSql(binder, SpiQuery.Type.ATTRIBUTE);
+      final DataReader dataReader = binder.createDataReader(request.getResultSet());
+      while (dataReader.next()) {
+        consumer.accept(scalarType.read(dataReader));
+      }
+      request.logSummary();
+
+    } catch (Exception e) {
+      throw new PersistenceException(errMsg(e.getMessage(), request.getSql()), e);
+
+    } finally {
+      request.close();
+    }
+  }
 }

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

@@ -254,6 +254,15 @@ public class DefaultRelationalQuery implements SpiSqlQuery {
     public List<T> findList() {
       return findSingleAttributeList(type);
     }
+
+    @Override
+    public void findEach(Consumer<T> consumer) {
+      scalarFindEach(type, consumer);
+    }
+  }
+
+  private <T> void scalarFindEach(Class<T> type, Consumer<T> consumer) {
+    server.findSingleAttributeEach(this, type, consumer);
   }
 
   private class Mapper<T> implements SqlQuery.TypeQuery<T> {
@@ -279,7 +288,7 @@ public class DefaultRelationalQuery implements SpiSqlQuery {
       return mapperFindList(mapper);
     }
 
-    //@Override
+    @Override
     public void findEach(Consumer<T> consumer) {
       mapperFindEach(mapper, consumer);
     }