#675 - ENH: Add SqlQuery.findEach() and SqlQuery.findEachWhile() methods ... to replace setListener()

src/main/java/com/avaje/ebean/EbeanServer.java

@@ -1192,6 +1192,25 @@ public interface EbeanServer {
    */
   List<SqlRow> findList(SqlQuery query, Transaction transaction);
 
+  /**
+   * Execute the SqlQuery 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(SqlQuery query, QueryEachConsumer<SqlRow> consumer, Transaction transaction);
+
+  /**
+   * Execute the SqlQuery 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(SqlQuery query, QueryEachWhileConsumer<SqlRow> consumer, Transaction transaction);
+
   /**
    * Execute the sql query returning a set of MapBean.
    * <p>

src/main/java/com/avaje/ebean/SqlQuery.java

@@ -61,6 +61,25 @@ public interface SqlQuery extends Serializable {
    */
   Map<?, SqlRow> findMap();
 
+  /**
+   * Execute the SqlQuery 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(QueryEachConsumer<SqlRow> consumer);
+
+  /**
+   * Execute the SqlQuery 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(QueryEachWhileConsumer<SqlRow> consumer);
+
   /**
    * Execute the query returning a single row or null.
    * <p>

src/main/java/com/avaje/ebeaninternal/server/core/DefaultServer.java

@@ -1404,6 +1404,30 @@ public final class DefaultServer implements SpiServer, SpiEbeanServer {
     return new SqlQueryFutureList(query, futureTask);
   }
 
+  @Override
+  public void findEach(SqlQuery query, QueryEachConsumer<SqlRow> consumer, Transaction transaction) {
+
+    RelationalQueryRequest request = new RelationalQueryRequest(this, relationalQueryEngine, query, transaction);
+    try {
+      request.initTransIfRequired();
+      request.findEach(consumer);
+    } finally {
+      request.endTransIfRequired();
+    }
+  }
+
+  @Override
+  public void findEachWhile(SqlQuery query, QueryEachWhileConsumer<SqlRow> consumer, Transaction transaction) {
+
+    RelationalQueryRequest request = new RelationalQueryRequest(this, relationalQueryEngine, query, transaction);
+    try {
+      request.initTransIfRequired();
+      request.findEachWhile(consumer);
+    } finally {
+      request.endTransIfRequired();
+    }
+  }
+
   public List<SqlRow> findList(SqlQuery query, Transaction t) {
 
     RelationalQueryRequest request = new RelationalQueryRequest(this, relationalQueryEngine, query, t);

src/main/java/com/avaje/ebeaninternal/server/core/RelationalQueryEngine.java

@@ -1,11 +1,25 @@
 package com.avaje.ebeaninternal.server.core;
 
 
+import com.avaje.ebean.QueryEachConsumer;
+import com.avaje.ebean.QueryEachWhileConsumer;
+import com.avaje.ebean.SqlRow;
+
 public interface RelationalQueryEngine {
 
-	/**
-	 * Find a list of beans using relational query.
-	 */
-	Object findMany(RelationalQueryRequest request);
+  /**
+   * Find a list of beans using relational query.
+   */
+  Object findMany(RelationalQueryRequest request);
+
+  /**
+   * Find each query using relational query.
+   */
+  void findEach(RelationalQueryRequest request, QueryEachConsumer<SqlRow> consumer);
+
+  /**
+   * Find each while query using relational query.
+   */
+  void findEach(RelationalQueryRequest request, QueryEachWhileConsumer<SqlRow> consumer);
 
 }

src/main/java/com/avaje/ebeaninternal/server/core/RelationalQueryRequest.java

@@ -1,114 +1,314 @@
 package com.avaje.ebeaninternal.server.core;
 
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
 import com.avaje.ebean.EbeanServer;
+import com.avaje.ebean.QueryEachConsumer;
+import com.avaje.ebean.QueryEachWhileConsumer;
 import com.avaje.ebean.SqlQuery;
 import com.avaje.ebean.SqlRow;
 import com.avaje.ebean.Transaction;
+import com.avaje.ebeaninternal.api.BindParams;
 import com.avaje.ebeaninternal.api.SpiEbeanServer;
 import com.avaje.ebeaninternal.api.SpiQuery;
 import com.avaje.ebeaninternal.api.SpiSqlQuery;
 import com.avaje.ebeaninternal.api.SpiTransaction;
+import com.avaje.ebeaninternal.server.lib.util.Str;
+import com.avaje.ebeaninternal.server.persist.Binder;
+import com.avaje.ebeaninternal.server.query.DefaultSqlRow;
+import com.avaje.ebeaninternal.server.transaction.TransactionManager;
+import com.avaje.ebeaninternal.server.util.BindParamsParser;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
 
 /**
  * Wraps the objects involved in executing a SqlQuery.
  */
 public final class RelationalQueryRequest {
 
-    private final SpiSqlQuery query;
+  private static final int GLOBAL_ROW_LIMIT = Integer.valueOf(System.getProperty("ebean.query.globallimit", "1000000"));
 
-    private final RelationalQueryEngine queryEngine;
+  private static final Logger logger = LoggerFactory.getLogger(RelationalQueryRequest.class);
 
-    private final SpiEbeanServer ebeanServer;
+  private final SpiSqlQuery query;
 
-    private SpiTransaction trans;
+  private final RelationalQueryEngine queryEngine;
 
-    private boolean createdTransaction;
+  private final SpiEbeanServer ebeanServer;
 
-    private SpiQuery.Type queryType;
+  private SpiTransaction trans;
 
-    /**
-     * Create the BeanFindRequest.
-     */
-    public RelationalQueryRequest(SpiEbeanServer server, RelationalQueryEngine engine, SqlQuery q, Transaction t) {
-        this.ebeanServer = server;
-        this.queryEngine = engine;
-        this.query = (SpiSqlQuery) q;
-        this.trans = (SpiTransaction) t;
+  private boolean createdTransaction;
+
+  private SpiQuery.Type queryType;
+
+  private String sql;
+
+  private ResultSet resultSet;
+
+  private int rowCount;
+
+  private String bindLog = "";
+
+  private String[] propertyNames;
+
+  private int estimateCapacity;
+
+  private PreparedStatement pstmt;
+
+  /**
+   * Create the BeanFindRequest.
+   */
+  public RelationalQueryRequest(SpiEbeanServer server, RelationalQueryEngine engine, SqlQuery q, Transaction t) {
+    this.ebeanServer = server;
+    this.queryEngine = engine;
+    this.query = (SpiSqlQuery) q;
+    this.trans = (SpiTransaction) t;
+  }
+
+  /**
+   * Create a transaction if none currently exists.
+   */
+  public void initTransIfRequired() {
+    if (trans == null) {
+      trans = ebeanServer.getCurrentServerTransaction();
+      if (trans == null || !trans.isActive()) {
+        // create a local readOnly transaction
+        trans = ebeanServer.createServerTransaction(false, -1);
+        createdTransaction = true;
+      }
     }
+  }
 
-    /**
-     * Create a transaction if none currently exists.
-     */
-    public void initTransIfRequired() {
-        if (trans == null) {
-            trans = ebeanServer.getCurrentServerTransaction();
-            if (trans == null || !trans.isActive()) {
-                // create a local readOnly transaction
-                trans = ebeanServer.createServerTransaction(false, -1);
-                createdTransaction = true;
-            }
+  /**
+   * End the transaction if it was locally created.
+   */
+  public void endTransIfRequired() {
+    if (createdTransaction) {
+      trans.commit();
+    }
+  }
+
+  public void findEach(QueryEachConsumer<SqlRow> consumer) {
+    queryType = SpiQuery.Type.ITERATE;
+    queryEngine.findEach(this, consumer);
+  }
+
+  public void findEachWhile(QueryEachWhileConsumer<SqlRow> consumer) {
+    queryType = SpiQuery.Type.ITERATE;
+    queryEngine.findEach(this, consumer);
+  }
+
+  @SuppressWarnings("unchecked")
+  public List<SqlRow> findList() {
+    queryType = SpiQuery.Type.LIST;
+    return (List<SqlRow>) queryEngine.findMany(this);
+  }
+
+  @SuppressWarnings("unchecked")
+  public Set<SqlRow> findSet() {
+    queryType = SpiQuery.Type.SET;
+    return (Set<SqlRow>) queryEngine.findMany(this);
+  }
+
+  @SuppressWarnings("unchecked")
+  public Map<?, SqlRow> findMap() {
+    queryType = SpiQuery.Type.MAP;
+    return (Map<?, SqlRow>) queryEngine.findMany(this);
+  }
+
+  /**
+   * Return the find that is to be performed.
+   */
+  public SpiSqlQuery getQuery() {
+    return query;
+  }
+
+  /**
+   * Return the type (List, Set or Map) that this fetch returns.
+   */
+  public SpiQuery.Type getQueryType() {
+    return queryType;
+  }
+
+  public EbeanServer getEbeanServer() {
+    return ebeanServer;
+  }
+
+  public SpiTransaction getTransaction() {
+    return trans;
+  }
+
+  public boolean isLogSql() {
+    return trans.isLogSql();
+  }
+
+  public boolean isLogSummary() {
+    return trans.isLogSummary();
+  }
+
+  private void setResultSet(ResultSet resultSet) throws SQLException {
+    this.resultSet = resultSet;
+    this.propertyNames = getPropertyNames();
+    // calculate the initialCapacity of the Map to reduce rehashing
+    float initCap = (propertyNames.length) / 0.7f;
+    this.estimateCapacity = (int) initCap + 1;
+  }
+
+  /**
+   * Build the list of property names.
+   */
+  private String[] getPropertyNames() throws SQLException {
+
+    ArrayList<String> propNames = new ArrayList<String>();
+    ResultSetMetaData metaData = resultSet.getMetaData();
+
+    int columnsPlusOne = metaData.getColumnCount() + 1;
+    for (int i = 1; i < columnsPlusOne; i++) {
+      propNames.add(metaData.getColumnLabel(i));
+    }
+    return propNames.toArray(new String[propNames.size()]);
+  }
+
+  /**
+   * Return the bindLog for this request.
+   */
+  public String getBindLog() {
+    return bindLog;
+  }
+
+  /**
+   * Return true if we can navigate to the next row.
+   */
+  public boolean next() throws SQLException {
+    rowCount++;
+    return resultSet.next();
+  }
+
+  /**
+   * Close the underlying resources.
+   */
+  public void close() {
+    try {
+      if (resultSet != null) {
+        resultSet.close();
+      }
+    } catch (SQLException e) {
+      logger.error(null, e);
+    }
+    try {
+      if (pstmt != null) {
+        pstmt.close();
+      }
+    } catch (SQLException e) {
+      logger.error(null, e);
+    }
+  }
+
+  /**
+   * Read and return the next SqlRow.
+   */
+  public SqlRow createNewRow(String dbTrueValue) throws SQLException {
+
+    SqlRow sqlRow = new DefaultSqlRow(estimateCapacity, 0.75f, dbTrueValue);
+
+    int index = 0;
+    for (int i = 0; i < propertyNames.length; i++) {
+      index++;
+      Object value = resultSet.getObject(index);
+      sqlRow.set(propertyNames[i], value);
+    }
+    return sqlRow;
+  }
+
+  /**
+   * Prepare the SQL taking into account named bind parameters.
+   */
+  private void prepareSql() {
+
+    String sql = query.getQuery();
+    BindParams bindParams = query.getBindParams();
+    if (!bindParams.isEmpty()) {
+      // convert any named parameters if required
+      sql = BindParamsParser.parse(bindParams, sql);
+    }
+    this.sql = sql;
+  }
+
+  /**
+   * Prepare and execute the SQL using the Binder.
+   */
+  public boolean executeSql(Binder binder) throws SQLException {
+
+    prepareSql();
+    synchronized (query) {
+      if (query.isCancelled()) {
+        logger.trace("Query already cancelled");
+        return false;
+      }
+
+      Connection conn = trans.getInternalConnection();
+
+      // synchronise for query.cancel() support
+      pstmt = conn.prepareStatement(sql);
+      if (query.getTimeout() > 0) {
+        pstmt.setQueryTimeout(query.getTimeout());
+      }
+      if (query.getBufferFetchSizeHint() > 0) {
+        pstmt.setFetchSize(query.getBufferFetchSizeHint());
+      }
+
+      BindParams bindParams = query.getBindParams();
+      if (!bindParams.isEmpty()) {
+        this.bindLog = binder.bind(bindParams, pstmt);
+      }
+
+      if (isLogSql()) {
+        String logSql = sql;
+        if (TransactionManager.SQL_LOGGER.isTraceEnabled()) {
+          logSql = Str.add(logSql, "; --bind(", bindLog, ")");
         }
-    }
+        trans.logSql(logSql);
+      }
 
-    /**
-     * End the transaction if it was locally created.
-     */
-    public void endTransIfRequired() {
-        if (createdTransaction) {
-            trans.commit();
-        }
+      setResultSet(pstmt.executeQuery());
+      return true;
     }
+  }
 
-    @SuppressWarnings("unchecked")
-    public List<SqlRow> findList() {
-        queryType = SpiQuery.Type.LIST;
-        return (List<SqlRow>) queryEngine.findMany(this);
-    }
+  /**
+   * Return true if the query has been cancelled.
+   */
+  public boolean isCancelled() {
+    return query.isCancelled();
+  }
 
-    @SuppressWarnings("unchecked")
-    public Set<SqlRow> findSet() {
-        queryType = SpiQuery.Type.SET;
-        return (Set<SqlRow>) queryEngine.findMany(this);
-    }
+  /**
+   * Return the maxRows allowed to fetch.
+   */
+  public int getMaxRows() {
+    return query.getMaxRows() >= 1 ? query.getMaxRows() : GLOBAL_ROW_LIMIT;
+  }
 
-    @SuppressWarnings("unchecked")
-    public Map<?, SqlRow> findMap() {
-        queryType = SpiQuery.Type.MAP;
-        return (Map<?, SqlRow>) queryEngine.findMany(this);
-    }
-
-    /**
-     * Return the find that is to be performed.
-     */
-    public SpiSqlQuery getQuery() {
-        return query;
-    }
-
-    /**
-     * Return the type (List, Set or Map) that this fetch returns.
-     */
-    public SpiQuery.Type getQueryType() {
-        return queryType;
-    }
-
-    public EbeanServer getEbeanServer() {
-        return ebeanServer;
-    }
-
-    public SpiTransaction getTransaction() {
-        return trans;
-    }
-
-    public boolean isLogSql() {
-        return trans.isLogSql();
-    }
-
-    public boolean isLogSummary() {
-        return trans.isLogSummary();
-    }
+  /**
+   * Return the SQL executed for this query.
+   */
+  public String getSql() {
+    return sql;
+  }
 
+  /**
+   * Return the rows read.
+   */
+  public int getRowCount() {
+    return rowCount - 1;
+  }
 }

src/main/java/com/avaje/ebeaninternal/server/query/DefaultRelationalQueryEngine.java

@@ -1,25 +1,19 @@
 package com.avaje.ebeaninternal.server.query;
 
+import com.avaje.ebean.QueryEachConsumer;
+import com.avaje.ebean.QueryEachWhileConsumer;
 import com.avaje.ebean.SqlQueryListener;
 import com.avaje.ebean.SqlRow;
-import com.avaje.ebean.bean.BeanCollection;
-import com.avaje.ebeaninternal.api.BindParams;
 import com.avaje.ebeaninternal.api.SpiSqlQuery;
-import com.avaje.ebeaninternal.api.SpiTransaction;
 import com.avaje.ebeaninternal.server.core.Message;
 import com.avaje.ebeaninternal.server.core.RelationalQueryEngine;
 import com.avaje.ebeaninternal.server.core.RelationalQueryRequest;
-import com.avaje.ebeaninternal.server.lib.util.Str;
 import com.avaje.ebeaninternal.server.persist.Binder;
-import com.avaje.ebeaninternal.server.transaction.TransactionManager;
-import com.avaje.ebeaninternal.server.type.DataBind;
-import com.avaje.ebeaninternal.server.util.BindParamsParser;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import javax.persistence.PersistenceException;
-import java.sql.*;
-import java.util.ArrayList;
+import java.sql.SQLException;
 
 /**
  * Perform native sql fetches.
@@ -28,7 +22,6 @@ public class DefaultRelationalQueryEngine implements RelationalQueryEngine {
 
   private static final Logger logger = LoggerFactory.getLogger(DefaultRelationalQueryEngine.class);
 
-  private static final int GLOBAL_ROW_LIMIT = Integer.valueOf(System.getProperty("ebean.query.globallimit", "1000000"));
 
   private final Binder binder;
 
@@ -39,108 +32,85 @@ public class DefaultRelationalQueryEngine implements RelationalQueryEngine {
     this.dbTrueValue = dbTrueValue == null ? "true" : dbTrueValue;
   }
 
-  public Object findMany(RelationalQueryRequest request) {
+  @Override
+  public void findEach(RelationalQueryRequest request, QueryEachWhileConsumer<SqlRow> consumer) {
 
-    SpiSqlQuery query = request.getQuery();
+    long startTime = System.currentTimeMillis();
+    try {
+      request.executeSql(binder);
+      while (request.next()) {
+        if (!consumer.accept(readRow(request))) {
+          break;
+        }
+      }
+      logSummary(request, startTime);
+
+    } catch (Exception e) {
+      throw new PersistenceException(Message.msg("fetch.error", e.getMessage(), request.getSql()), e);
+
+    } finally {
+      request.close();
+    }
+  }
+
+  @Override
+  public void findEach(RelationalQueryRequest request, QueryEachConsumer<SqlRow> consumer) {
 
     long startTime = System.currentTimeMillis();
 
-    SpiTransaction t = request.getTransaction();
-    Connection conn = t.getInternalConnection();
-    ResultSet rset = null;
-    PreparedStatement pstmt = null;
-
-    String sql = query.getQuery();
-
-    BindParams bindParams = query.getBindParams();
-
-    if (!bindParams.isEmpty()) {
-      // convert any named parameters if required
-      sql = BindParamsParser.parse(bindParams, sql);
-    }
-
     try {
+      request.executeSql(binder);
+      while (request.next()) {
+        consumer.accept(readRow(request));
+      }
+      logSummary(request, startTime);
 
-      String bindLog = "";
-      String[] propNames;
+    } catch (Exception e) {
+      throw new PersistenceException(Message.msg("fetch.error", e.getMessage(), request.getSql()), e);
 
-      synchronized (query) {
-        if (query.isCancelled()) {
-          logger.trace("Query already cancelled");
-          return null;
-        }
+    } finally {
+      request.close();
+    }
+  }
 
-        // synchronise for query.cancel() support
-        pstmt = conn.prepareStatement(sql);
+  public Object findMany(RelationalQueryRequest request) {
 
-        if (query.getTimeout() > 0) {
-          pstmt.setQueryTimeout(query.getTimeout());
-        }
-        if (query.getBufferFetchSizeHint() > 0) {
-          pstmt.setFetchSize(query.getBufferFetchSizeHint());
-        }
-
-        if (!bindParams.isEmpty()) {
-          bindLog = binder.bind(bindParams, pstmt);
-        }
-
-        if (request.isLogSql()) {
-          String logSql = sql;
-          if (TransactionManager.SQL_LOGGER.isTraceEnabled()) {
-            logSql = Str.add(logSql, "; --bind(", bindLog, ")");
-          }
-          t.logSql(logSql);
-        }
-
-        rset = pstmt.executeQuery();
-
-        propNames = getPropertyNames(rset);
+    long startTime = System.currentTimeMillis();
+    try {
+      if (!request.executeSql(binder)) {
+        return null;
       }
 
-      // calculate the initialCapacity of the Map to reduce
-      // rehashing for queries with 12+ columns
-      float initCap = (propNames.length) / 0.7f;
-      int estimateCapacity = (int) initCap + 1;
-
-      // determine the maxRows limit
-      int maxRows = GLOBAL_ROW_LIMIT;
-      if (query.getMaxRows() >= 1) {
-        maxRows = query.getMaxRows();
-      }
+      int maxRows = request.getMaxRows();
 
       int loadRowCount = 0;
 
-      SqlQueryListener listener = query.getListener();
-
       BeanCollectionWrapper wrapper = new BeanCollectionWrapper(request);
       boolean isMap = wrapper.isMap();
-      String mapKey = query.getMapKey();
+      String mapKey = request.getQuery().getMapKey();
 
-      SqlRow bean = null;
+      SpiSqlQuery query = request.getQuery();
+      SqlQueryListener listener = query.getListener();
 
-      while (rset.next()) {
+      while (request.next()) {
+        SqlRow bean;
         synchronized (query) {
-          // synchronise for query.cancel() support
-          if (!query.isCancelled()) {
-            bean = readRow(rset, propNames, estimateCapacity);
+          if (request.isCancelled()) {
+            break;
           }
+          bean = readRow(request);
         }
         if (bean != null) {
-          // bean can be null if query cancelled
           if (listener != null) {
             listener.process(bean);
-
           } else {
             if (isMap) {
-              Object keyValue = bean.get(mapKey);
-              wrapper.addToMap(bean, keyValue);
+              wrapper.addToMap(bean, bean.get(mapKey));
             } else {
               wrapper.addToCollection(bean);
             }
           }
-
           loadRowCount++;
-
           if (loadRowCount == maxRows) {
             // break, as we have hit the max rows to fetch...
             break;
@@ -148,84 +118,35 @@ public class DefaultRelationalQueryEngine implements RelationalQueryEngine {
         }
       }
 
-      BeanCollection<?> beanColl = wrapper.getBeanCollection();
+      logSummary(request, startTime);
 
-      if (request.isLogSummary()) {
-        long exeTime = System.currentTimeMillis() - startTime;
-        String msg = "SqlQuery  rows[" + loadRowCount + "] time[" + exeTime + "] bind[" + bindLog + "]";
-        t.logSummary(msg);
-      }
-
-      if (query.isCancelled()) {
-        logger.debug("Query was cancelled during execution rows:" + loadRowCount);
-      }
-
-      return beanColl;
+      return wrapper.getBeanCollection();
 
     } catch (Exception e) {
-      String m = Message.msg("fetch.error", e.getMessage(), sql);
-      throw new PersistenceException(m, e);
+      throw new PersistenceException(Message.msg("fetch.error", e.getMessage(), request.getSql()), e);
 
     } finally {
-      try {
-        if (rset != null) {
-          rset.close();
-        }
-      } catch (SQLException e) {
-        logger.error(null, e);
-      }
-      try {
-        if (pstmt != null) {
-          pstmt.close();
-        }
-      } catch (SQLException e) {
-        logger.error(null, e);
-      }
+      request.close();
     }
   }
 
-  /**
-   * Build the list of property names.
-   */
-  protected String[] getPropertyNames(ResultSet rset) throws SQLException {
+  private void logSummary(RelationalQueryRequest request, long startTime) {
 
-    ArrayList<String> propNames = new ArrayList<String>();
-
-    ResultSetMetaData rsmd = rset.getMetaData();
-
-    int columnsPlusOne = rsmd.getColumnCount() + 1;
-
-
-    for (int i = 1; i < columnsPlusOne; i++) {
-      String columnName = rsmd.getColumnLabel(i);
-      // will convert columnName to lower case
-      propNames.add(columnName);
+    if (request.isLogSummary()) {
+      long exeTime = System.currentTimeMillis() - startTime;
+      request.getTransaction().logSummary("SqlQuery  rows[" + request.getRowCount() + "] time[" + exeTime + "] bind[" + request.getBindLog() + "]");
     }
 
-    return propNames.toArray(new String[propNames.size()]);
+    if (request.isCancelled()) {
+      logger.debug("Query was cancelled during execution rows: {}", request.getRowCount());
+    }
   }
 
   /**
    * Read the row from the ResultSet and return as a MapBean.
    */
-  protected SqlRow readRow(ResultSet rset, String[] propNames, int initialCapacity) throws SQLException {
-
-    // by default a map will rehash on the 12th entry
-    // it will be pretty common to have 12 or more entries so
-    // to reduce rehashing I am trying to estimate a good
-    // initial capacity for the MapBean to use.
-    SqlRow bean = new DefaultSqlRow(initialCapacity, 0.75f, dbTrueValue);
-
-    int index = 0;
-
-    for (int i = 0; i < propNames.length; i++) {
-      index++;
-      Object value = rset.getObject(index);
-      bean.set(propNames[i], value);
-    }
-
-    return bean;
-
+  protected SqlRow readRow(RelationalQueryRequest request) throws SQLException {
+    return request.createNewRow(dbTrueValue);
   }
 
 }

src/main/java/com/avaje/ebeaninternal/server/querydefn/DefaultRelationalQuery.java

@@ -9,6 +9,8 @@ import java.util.Set;
 import javax.persistence.PersistenceException;
 
 import com.avaje.ebean.EbeanServer;
+import com.avaje.ebean.QueryEachConsumer;
+import com.avaje.ebean.QueryEachWhileConsumer;
 import com.avaje.ebean.SqlFutureList;
 import com.avaje.ebean.SqlQueryListener;
 import com.avaje.ebean.SqlRow;
@@ -73,6 +75,16 @@ public class DefaultRelationalQuery implements SpiSqlQuery {
     return this;
   }
 
+  @Override
+  public void findEach(QueryEachConsumer<SqlRow> consumer) {
+    server.findEach(this, consumer, null);
+  }
+
+  @Override
+  public void findEachWhile(QueryEachWhileConsumer<SqlRow> consumer) {
+    server.findEachWhile(this, consumer, null);
+  }
+
   public List<SqlRow> findList() {
     return server.findList(this, null);
   }

src/test/java/com/avaje/ebeaninternal/api/TDSpiEbeanServer.java

@@ -540,6 +540,14 @@ public class TDSpiEbeanServer implements SpiEbeanServer {
     return null;
   }
 
+  @Override
+  public void findEach(SqlQuery query, QueryEachConsumer<SqlRow> consumer, Transaction transaction) {
+  }
+
+  @Override
+  public void findEachWhile(SqlQuery query, QueryEachWhileConsumer<SqlRow> consumer, Transaction transaction) {
+  }
+
   @Override
   public Set<SqlRow> findSet(SqlQuery query, Transaction transaction) {
     return null;

src/test/java/com/avaje/tests/query/sqlquery/SqlQueryTests.java

@@ -1,29 +1,81 @@
 package com.avaje.tests.query.sqlquery;
 
 import com.avaje.ebean.Ebean;
+import com.avaje.ebean.QueryEachConsumer;
+import com.avaje.ebean.QueryEachWhileConsumer;
 import com.avaje.ebean.SqlQuery;
 import com.avaje.ebean.SqlQueryListener;
 import com.avaje.ebean.SqlRow;
+import com.avaje.tests.model.basic.Order;
 import com.avaje.tests.model.basic.ResetBasicData;
 import org.junit.Test;
 
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.Assert.assertEquals;
+
 public class SqlQueryTests {
 
   @Test
   public void setListener() {
 
-
     ResetBasicData.reset();
 
+    int expectedRows = Ebean.find(Order.class).findRowCount();
+
+    final AtomicInteger count = new AtomicInteger();
+
     SqlQuery sqlQuery = Ebean.createSqlQuery("select * from o_order");
     sqlQuery.setListener(new SqlQueryListener() {
       @Override
       public void process(SqlRow bean) {
         System.out.println("process row "+bean);
+        count.incrementAndGet();
       }
     });
     // returns an empty list
     sqlQuery.findList();
-
+    assertEquals(expectedRows, count.get());
   }
+
+  @Test
+  public void findEach() {
+
+    ResetBasicData.reset();
+
+    int expectedRows = Ebean.find(Order.class).findRowCount();
+
+    final AtomicInteger count = new AtomicInteger();
+
+    SqlQuery sqlQuery = Ebean.createSqlQuery("select * from o_order");
+    sqlQuery.findEach(new QueryEachConsumer<SqlRow>() {
+      @Override
+      public void accept(SqlRow bean) {
+        count.incrementAndGet();
+      }
+    });
+
+    assertEquals(expectedRows, count.get());
+  }
+
+  @Test
+  public void findEachWhile() {
+
+    ResetBasicData.reset();
+
+    final AtomicInteger count = new AtomicInteger();
+
+    SqlQuery sqlQuery = Ebean.createSqlQuery("select * from o_order order by id");
+    sqlQuery.findEachWhile(new QueryEachWhileConsumer<SqlRow>() {
+      @Override
+      public boolean accept(SqlRow bean) {
+        count.incrementAndGet();
+        Integer id = bean.getInteger("id");
+        return id.intValue() < 3;
+      }
+    });
+
+    assertEquals(3, count.get());
+  }
+
 }