Queries are cancelable

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

@@ -0,0 +1,26 @@
+package io.ebeaninternal.api;
+
+import javax.persistence.PersistenceException;
+
+import io.ebean.CancelableQuery;
+
+/**
+ * Cancellable query, that has a delegate.
+ *
+ * @author Roland Praml, FOCONIS AG
+ *
+ */
+public interface SpiCancelableQuery extends CancelableQuery {
+
+  /**
+   * Checks if the query was cancelled.
+   * @throws PersistenceException if query was cancelled.
+   */
+  void checkCancelled();
+
+  /**
+   * Set the underlying cancelable query (with the PreparedStatement).
+   */
+  void setCancelableQuery(CancelableQuery cancelableQuery);
+
+}

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

@@ -17,7 +17,6 @@ import io.ebeaninternal.server.core.SpiOrmQueryRequest;
 import io.ebeaninternal.server.deploy.BeanDescriptor;
 import io.ebeaninternal.server.deploy.BeanPropertyAssocMany;
 import io.ebeaninternal.server.deploy.TableJoin;
-import io.ebeaninternal.server.query.CancelableQuery;
 import io.ebeaninternal.server.querydefn.NaturalKeyBindParam;
 import io.ebeaninternal.server.querydefn.OrmQueryDetail;
 import io.ebeaninternal.server.querydefn.OrmQueryProperties;
@@ -31,7 +30,7 @@ import java.util.Set;
 /**
  * Object Relational query - Internal extension to Query object.
  */
-public interface SpiQuery<T> extends Query<T>, SpiQueryFetch, TxnProfileEventCodes {
+public interface SpiQuery<T> extends Query<T>, SpiQueryFetch, TxnProfileEventCodes, SpiCancelableQuery {
 
   enum Mode {
     NORMAL(false), LAZYLOAD_MANY(false), LAZYLOAD_BEAN(true), REFRESH_BEAN(true);
@@ -847,16 +846,6 @@ public interface SpiQuery<T> extends Query<T>, SpiQueryFetch, TxnProfileEventCod
    */
   ReadEvent getFutureFetchAudit();
 
-  /**
-   * Set the underlying cancelable query (with the PreparedStatement).
-   */
-  void setCancelableQuery(CancelableQuery cancelableQuery);
-
-  /**
-   * Return true if this query has been cancelled.
-   */
-  boolean isCancelled();
-
   /**
    * Return the base table to use if user defined on the query.
    */

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

@@ -3,7 +3,7 @@ package io.ebeaninternal.api;
 /**
  * SQL query binding (for SqlQuery and DtoQuery).
  */
-public interface SpiSqlBinding {
+public interface SpiSqlBinding extends SpiCancelableQuery {
 
   /**
    * Return the named or positioned parameters.

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

@@ -1,5 +1,6 @@
 package io.ebeaninternal.server.core;
 
+import io.ebean.CancelableQuery;
 import io.ebean.Transaction;
 import io.ebean.util.JdbcClose;
 import io.ebeaninternal.api.*;
@@ -12,11 +13,14 @@ import java.sql.Connection;
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.util.concurrent.locks.ReentrantLock;
+
+import javax.persistence.PersistenceException;
 
 /**
  * Wraps the objects involved in executing a SQL / Relational Query.
  */
-public abstract class AbstractSqlQueryRequest {
+public abstract class AbstractSqlQueryRequest implements CancelableQuery {
 
   protected final SpiSqlBinding query;
 
@@ -36,6 +40,8 @@ public abstract class AbstractSqlQueryRequest {
 
   protected long startNano;
 
+  private final ReentrantLock lock = new ReentrantLock();
+  
   /**
    * Create the BeanFindRequest.
    */
@@ -43,6 +49,7 @@ public abstract class AbstractSqlQueryRequest {
     this.server = server;
     this.query = query;
     this.transaction = (SpiTransaction) t;
+    this.query.setCancelableQuery(this);
   }
 
   /**
@@ -137,24 +144,30 @@ public abstract class AbstractSqlQueryRequest {
   }
 
   protected void executeAsSql(Binder binder) throws SQLException {
-    prepareSql();
-    Connection conn = transaction.getInternalConnection();
-    pstmt = conn.prepareStatement(sql);
-    if (query.getTimeout() > 0) {
-      pstmt.setQueryTimeout(query.getTimeout());
+    lock.lock();
+    try {
+      query.checkCancelled();
+      prepareSql();
+      Connection conn = transaction.getInternalConnection();
+      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, conn);
+      }
+      if (isLogSql()) {
+        transaction.logSql(Str.add(TrimLogSql.trim(sql), "; --bind(", bindLog, ")"));
+      }
+    } finally {
+      lock.unlock();
     }
-    if (query.getBufferFetchSizeHint() > 0) {
-      pstmt.setFetchSize(query.getBufferFetchSizeHint());
-    }
-    BindParams bindParams = query.getBindParams();
-    if (!bindParams.isEmpty()) {
-      this.bindLog = binder.bind(bindParams, pstmt, conn);
-    }
-    if (isLogSql()) {
-      transaction.logSql(Str.add(TrimLogSql.trim(sql), "; --bind(", bindLog, ")"));
-    }
-
     setResultSet(pstmt.executeQuery(), null);
+    query.checkCancelled();
   }
 
   /**
@@ -164,4 +177,13 @@ public abstract class AbstractSqlQueryRequest {
     return sql;
   }
 
+  @Override
+  public void cancel() {
+    lock.lock();
+    try {
+      JdbcClose.cancel(pstmt);
+    } finally {
+      lock.unlock();
+    }
+  }
 }

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

@@ -1382,7 +1382,7 @@ public final class DefaultServer implements SpiServer, SpiEbeanServer {
   @Nonnull
   @Override
   public <T> FutureList<T> findFutureList(Query<T> query, Transaction t) {
-    SpiQuery<T> spiQuery = (SpiQuery<T>) query;
+    SpiQuery<T> spiQuery = (SpiQuery<T>) query.copy();
     spiQuery.setFutureFetch(true);
     // FutureList query always run in it's own persistence content
     spiQuery.setPersistenceContext(new DefaultPersistenceContext());

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

@@ -53,6 +53,7 @@ public final class DtoQueryRequest<T> extends AbstractSqlQueryRequest {
       ormQuery.setType(type);
       ormQuery.setManualId();
 
+      query.setCancelableQuery(ormQuery);
       // execute the underlying ORM query returning the ResultSet
       SpiResultSet result = server.findResultSet(ormQuery, transaction);
       this.pstmt = result.getStatement();
@@ -117,6 +118,7 @@ public final class DtoQueryRequest<T> extends AbstractSqlQueryRequest {
   }
 
   public boolean next() throws SQLException {
+    query.checkCancelled();
     return dataReader.next();
   }
 

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

@@ -1,6 +1,7 @@
 package io.ebeaninternal.server.core;
 
 import io.ebean.CacheMode;
+import io.ebean.CancelableQuery;
 import io.ebean.OrderBy;
 import io.ebean.PersistenceContextScope;
 import io.ebean.QueryIterator;
@@ -35,7 +36,6 @@ import io.ebeaninternal.server.deploy.DeployPropertyParserMap;
 import io.ebeaninternal.server.el.ElPropertyValue;
 import io.ebeaninternal.server.loadcontext.DLoadContext;
 import io.ebeaninternal.server.query.CQueryPlan;
-import io.ebeaninternal.server.query.CancelableQuery;
 import io.ebeaninternal.server.transaction.DefaultPersistenceContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

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

@@ -139,6 +139,7 @@ public final class RelationalQueryRequest extends AbstractSqlQueryRequest {
 
   @Override
   public boolean next() throws SQLException {
+    query.checkCancelled();
     if (!resultSet.next()) {
       return false;
     } else {

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

@@ -1,5 +1,6 @@
 package io.ebeaninternal.server.query;
 
+import io.ebean.CancelableQuery;
 import io.ebean.QueryIterator;
 import io.ebean.Version;
 import io.ebean.bean.*;
@@ -139,8 +140,6 @@ public class CQuery<T> implements DbReadContext, CancelableQuery, SpiProfileTran
    */
   private PreparedStatement pstmt;
 
-  private boolean cancelled;
-
   private String bindLog;
 
   private final CQueryPlan queryPlan;
@@ -272,15 +271,7 @@ public class CQuery<T> implements DbReadContext, CancelableQuery, SpiProfileTran
   public void cancel() {
     lock.lock();
     try {
-      this.cancelled = true;
-      if (pstmt != null) {
-        try {
-          logger.debug("Cancelling query");
-          pstmt.cancel();
-        } catch (SQLException e) {
-          throw new PersistenceException("Error cancelling query", e);
-        }
-      }
+      JdbcClose.cancel(pstmt);
     } finally {
       lock.unlock();
     }
@@ -312,9 +303,8 @@ public class CQuery<T> implements DbReadContext, CancelableQuery, SpiProfileTran
   ResultSet prepareResultSet(boolean forwardOnlyHint) throws SQLException {
     lock.lock();
     try {
-      if (cancelled) {
-        throw new SQLException("Query cancelled");
-      }
+      // cancelled before we started
+      query.checkCancelled();
       startNano = System.nanoTime();
       SpiTransaction t = request.getTransaction();
       profileOffset = t.profileOffset();
@@ -342,10 +332,12 @@ public class CQuery<T> implements DbReadContext, CancelableQuery, SpiProfileTran
         pstmt.setFetchSize(query.getBufferFetchSizeHint());
       }
       bindLog = predicates.bind(queryPlan.bindEncryptedProperties(pstmt, conn));
-      return pstmt.executeQuery();
     } finally {
       lock.unlock();
     }
+    ResultSet ret = pstmt.executeQuery();
+    query.checkCancelled();
+    return ret;
   }
 
   /**
@@ -491,7 +483,8 @@ public class CQuery<T> implements DbReadContext, CancelableQuery, SpiProfileTran
   boolean hasNext() throws SQLException {
     lock.lock();
     try {
-      if (noMoreRows || cancelled) {
+      query.checkCancelled();
+      if (noMoreRows) {
         return false;
       }
       if (hasNextCache) {

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

@@ -66,11 +66,13 @@ public class CQueryEngine {
 
   public <T> int delete(OrmQueryRequest<T> request) {
     CQueryUpdate query = queryBuilder.buildUpdateQuery(true, request);
+    request.setCancelableQuery(query);
     return executeUpdate(request, query);
   }
 
   public <T> int update(OrmQueryRequest<T> request) {
     CQueryUpdate query = queryBuilder.buildUpdateQuery(false, request);
+    request.setCancelableQuery(query);
     return executeUpdate(request, query);
   }
 
@@ -97,6 +99,7 @@ public class CQueryEngine {
   public <A> List<A> findSingleAttributeList(OrmQueryRequest<?> request) {
 
     CQueryFetchSingleAttribute rcQuery = queryBuilder.buildFetchAttributeQuery(request);
+    request.setCancelableQuery(rcQuery);
     return findAttributeList(request, rcQuery);
   }
 
@@ -151,6 +154,7 @@ public class CQueryEngine {
   public <A> List<A> findIds(OrmQueryRequest<?> request) {
 
     CQueryFetchSingleAttribute rcQuery = queryBuilder.buildFetchIdsQuery(request);
+    request.setCancelableQuery(rcQuery);
     return findAttributeList(request, rcQuery);
   }
 
@@ -164,6 +168,7 @@ public class CQueryEngine {
   public <T> int findCount(OrmQueryRequest<T> request) {
 
     CQueryRowCount rcQuery = queryBuilder.buildRowCountQuery(request);
+    request.setCancelableQuery(rcQuery);
     try {
 
       int count = rcQuery.findCount();
@@ -235,8 +240,10 @@ public class CQueryEngine {
 
     } catch (SQLException e) {
       try {
+        PersistenceException pex = cquery.createPersistenceException(e);
+        // create exception before closing connection
         cquery.close();
-        throw cquery.createPersistenceException(e);
+        throw pex;
       } finally {
         request.rollbackTransIfRequired();
       }
@@ -259,6 +266,7 @@ public class CQueryEngine {
     // order by lower sys period desc
     query.order().desc(sysPeriodLower);
     CQuery<T> cquery = queryBuilder.buildQuery(request);
+    request.setCancelableQuery(cquery);
     try {
       cquery.prepareBindExecuteQuery();
       if (request.isLogSql()) {
@@ -327,6 +335,7 @@ public class CQueryEngine {
    */
   public <T> SpiResultSet findResultSet(OrmQueryRequest<T> request) {
     CQuery<T> cquery = queryBuilder.buildQuery(request);
+    request.setCancelableQuery(cquery);
     try {
       boolean fwdOnly;
       if (request.isFindIterate()) {
@@ -411,6 +420,7 @@ public class CQueryEngine {
     EntityBean bean = null;
 
     CQuery<T> cquery = queryBuilder.buildQuery(request);
+    request.setCancelableQuery(cquery);
 
     try {
       cquery.prepareBindExecuteQuery();

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

@@ -1,5 +1,6 @@
 package io.ebeaninternal.server.query;
 
+import io.ebean.CancelableQuery;
 import io.ebean.CountedValue;
 import io.ebean.core.type.ScalarDataReader;
 import io.ebean.util.JdbcClose;
@@ -18,11 +19,12 @@ import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.locks.ReentrantLock;
 
 /**
  * Base compiled query request for single attribute queries.
  */
-class CQueryFetchSingleAttribute implements SpiProfileTransactionEvent {
+class CQueryFetchSingleAttribute implements SpiProfileTransactionEvent, CancelableQuery {
 
   private static final Logger logger = LoggerFactory.getLogger(CQueryFetchSingleAttribute.class);
 
@@ -65,6 +67,8 @@ class CQueryFetchSingleAttribute implements SpiProfileTransactionEvent {
   private final boolean containsCounts;
 
   private long profileOffset;
+  
+  private final ReentrantLock lock = new ReentrantLock();
 
   /**
    * Create the Sql select based on the request.
@@ -147,21 +151,27 @@ class CQueryFetchSingleAttribute implements SpiProfileTransactionEvent {
   }
 
   private void prepareExecute() throws SQLException {
-
-    SpiTransaction t = getTransaction();
-    profileOffset = t.profileOffset();
-    Connection conn = t.getInternalConnection();
-    pstmt = conn.prepareStatement(sql);
-
-    if (query.getBufferFetchSizeHint() > 0) {
-      pstmt.setFetchSize(query.getBufferFetchSizeHint());
+    lock.lock();
+    try {
+      query.checkCancelled();
+      SpiTransaction t = getTransaction();
+      profileOffset = t.profileOffset();
+      Connection conn = t.getInternalConnection();
+      pstmt = conn.prepareStatement(sql);
+  
+      if (query.getBufferFetchSizeHint() > 0) {
+        pstmt.setFetchSize(query.getBufferFetchSizeHint());
+      }
+      if (query.getTimeout() > 0) {
+        pstmt.setQueryTimeout(query.getTimeout());
+      }
+  
+      bindLog = predicates.bind(pstmt, conn);
+    } finally {
+      lock.unlock();
     }
-    if (query.getTimeout() > 0) {
-      pstmt.setQueryTimeout(query.getTimeout());
-    }
-
-    bindLog = predicates.bind(pstmt, conn);
     dataReader = new RsetDataReader(request.getDataTimeZone(), pstmt.executeQuery());
+    query.checkCancelled();
   }
 
   /**
@@ -194,4 +204,14 @@ class CQueryFetchSingleAttribute implements SpiProfileTransactionEvent {
   Set<String> getDependentTables() {
     return queryPlan.getDependentTables();
   }
+
+  @Override
+  public void cancel() {
+    lock.lock();
+    try {
+      JdbcClose.cancel(pstmt);
+    } finally {
+      lock.unlock();
+    }
+  }
 }

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

@@ -1,5 +1,6 @@
 package io.ebeaninternal.server.query;
 
+import io.ebean.CancelableQuery;
 import io.ebean.util.JdbcClose;
 import io.ebeaninternal.api.SpiProfileTransactionEvent;
 import io.ebeaninternal.api.SpiQuery;
@@ -13,11 +14,12 @@ import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.Set;
+import java.util.concurrent.locks.ReentrantLock;
 
 /**
  * Executes the select row count query.
  */
-class CQueryRowCount implements SpiProfileTransactionEvent {
+class CQueryRowCount implements SpiProfileTransactionEvent, CancelableQuery {
 
   private final CQueryPlan queryPlan;
 
@@ -57,6 +59,8 @@ class CQueryRowCount implements SpiProfileTransactionEvent {
   private int rowCount;
 
   private long profileOffset;
+  
+  private final ReentrantLock lock = new ReentrantLock();
 
   /**
    * Create the Sql select based on the request.
@@ -110,14 +114,22 @@ class CQueryRowCount implements SpiProfileTransactionEvent {
       SpiTransaction t = getTransaction();
       profileOffset = t.profileOffset();
       Connection conn = t.getInternalConnection();
-      pstmt = conn.prepareStatement(sql);
+      lock.lock();
+      try {
+        query.checkCancelled();
+        pstmt = conn.prepareStatement(sql);
 
-      if (query.getTimeout() > 0) {
-        pstmt.setQueryTimeout(query.getTimeout());
+        if (query.getTimeout() > 0) {
+          pstmt.setQueryTimeout(query.getTimeout());
+        }
+
+        bindLog = predicates.bind(pstmt, conn);
+      } finally {
+        lock.unlock();
       }
-
-      bindLog = predicates.bind(pstmt, conn);
       rset = pstmt.executeQuery();
+      query.checkCancelled();
+      
       if (!rset.next()) {
         throw new PersistenceException("Expecting 1 row but got none?");
       }
@@ -161,4 +173,14 @@ class CQueryRowCount implements SpiProfileTransactionEvent {
   Set<String> getDependentTables() {
     return queryPlan.getDependentTables();
   }
+
+  @Override
+  public void cancel() {
+    lock.lock();
+    try {
+      JdbcClose.cancel(pstmt);
+    } finally {
+      lock.unlock();
+    }
+  }
 }

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

@@ -1,5 +1,6 @@
 package io.ebeaninternal.server.query;
 
+import io.ebean.CancelableQuery;
 import io.ebean.util.JdbcClose;
 import io.ebeaninternal.api.SpiProfileTransactionEvent;
 import io.ebeaninternal.api.SpiQuery;
@@ -10,11 +11,12 @@ import io.ebeaninternal.server.deploy.BeanDescriptor;
 import java.sql.Connection;
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
+import java.util.concurrent.locks.ReentrantLock;
 
 /**
- * Executes the delete query.
+ * Executes the update query.
  */
-class CQueryUpdate implements SpiProfileTransactionEvent {
+class CQueryUpdate implements SpiProfileTransactionEvent, CancelableQuery {
 
   private final CQueryPlan queryPlan;
 
@@ -45,6 +47,8 @@ class CQueryUpdate implements SpiProfileTransactionEvent {
 
   private long profileOffset;
 
+  private final ReentrantLock lock = new ReentrantLock();
+  
   /**
    * Create the Sql select based on the request.
    */
@@ -82,15 +86,22 @@ class CQueryUpdate implements SpiProfileTransactionEvent {
       SpiTransaction t = getTransaction();
       profileOffset = t.profileOffset();
       Connection conn = t.getInternalConnection();
-      pstmt = conn.prepareStatement(sql);
+      lock.lock();
+      try {
+        query.checkCancelled();
+        pstmt = conn.prepareStatement(sql);
 
-      if (query.getTimeout() > 0) {
-        pstmt.setQueryTimeout(query.getTimeout());
+        if (query.getTimeout() > 0) {
+          pstmt.setQueryTimeout(query.getTimeout());
+        }
+
+        bindLog = predicates.bind(pstmt, conn);
+      } finally {
+        lock.unlock();
       }
-
-      bindLog = predicates.bind(pstmt, conn);
       rowCount = pstmt.executeUpdate();
-
+      query.checkCancelled();
+      
       long executionTimeMicros = (System.nanoTime() - startNano) / 1000L;
       request.slowQueryCheck(executionTimeMicros, rowCount);
       if (queryPlan.executionTime(executionTimeMicros)) {
@@ -122,4 +133,14 @@ class CQueryUpdate implements SpiProfileTransactionEvent {
       .profileStream()
       .addQueryEvent(query.profileEventId(), profileOffset, desc.getName(), rowCount, query.getProfileId());
   }
+
+  @Override
+  public void cancel() {
+    lock.lock();
+    try {
+      JdbcClose.cancel(pstmt);
+    } finally {
+      lock.unlock();
+    }
+  }
 }

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

@@ -29,7 +29,7 @@ public class DtoQueryEngine {
       }
       return rows;
 
-    } catch (Throwable e) {
+    } catch (SQLException e) {
       throw new PersistenceException(errMsg(e.getMessage(), request.getSql()), e);
     } finally {
       request.close();
@@ -51,7 +51,7 @@ public class DtoQueryEngine {
       while (request.next()) {
         consumer.accept(request.readNextBean());
       }
-    } catch (Exception e) {
+    } catch (SQLException e) {
       throw new PersistenceException(errMsg(e.getMessage(), request.getSql()), e);
     } finally {
       request.close();
@@ -88,9 +88,8 @@ public class DtoQueryEngine {
           break;
         }
       }
-    } catch (Exception e) {
+    } catch (SQLException e) {
       throw new PersistenceException(errMsg(e.getMessage(), request.getSql()), e);
-
     } finally {
       request.close();
     }

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

@@ -0,0 +1,56 @@
+package io.ebeaninternal.server.querydefn;
+
+import java.util.concurrent.locks.ReentrantLock;
+
+import javax.persistence.PersistenceException;
+
+import io.ebean.CancelableQuery;
+import io.ebeaninternal.api.SpiCancelableQuery;
+
+/**
+ * Common code for Dto/Orm/RelationalQuery
+ *
+ * @author Roland Praml, FOCONIS AG
+ *
+ */
+public class AbstractQuery implements SpiCancelableQuery {
+
+  private boolean cancelled;
+
+  private CancelableQuery cancelableQuery;
+
+  private final ReentrantLock lock = new ReentrantLock();
+  
+  @Override
+  public void cancel() {
+    lock.lock();
+    try {
+      if (!cancelled) {
+        cancelled = true;
+        if (cancelableQuery != null) {
+          cancelableQuery.cancel();
+        }
+      }
+    } finally {
+      lock.unlock();
+    }
+  }
+
+  @Override
+  public void checkCancelled() {
+    if (cancelled) {
+      throw new PersistenceException("Query was cancelled");
+    }
+  }
+
+  @Override
+  public void setCancelableQuery(CancelableQuery cancelableQuery) {
+    lock.lock();
+    try {
+      checkCancelled();
+      this.cancelableQuery = cancelableQuery;
+    } finally {
+      lock.unlock();
+    }
+  }
+}

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

@@ -21,7 +21,7 @@ import java.util.stream.Stream;
 /**
  * Default implementation of DtoQuery.
  */
-public class DefaultDtoQuery<T> implements SpiDtoQuery<T> {
+public class DefaultDtoQuery<T> extends AbstractQuery implements SpiDtoQuery<T> {
 
   private final SpiEbeanServer server;
 

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

@@ -58,7 +58,6 @@ import io.ebeaninternal.server.deploy.TableJoin;
 import io.ebeaninternal.server.expression.DefaultExpressionList;
 import io.ebeaninternal.server.expression.IdInExpression;
 import io.ebeaninternal.server.expression.SimpleExpression;
-import io.ebeaninternal.server.query.CancelableQuery;
 import io.ebeaninternal.server.query.NativeSqlQueryPlanKey;
 import io.ebeaninternal.server.rawsql.SpiRawSql;
 import io.ebeaninternal.server.transaction.ExternalJdbcTransaction;
@@ -81,7 +80,7 @@ import java.util.stream.Stream;
 /**
  * Default implementation of an Object Relational query.
  */
-public class DefaultOrmQuery<T> implements SpiQuery<T> {
+public class DefaultOrmQuery<T> extends AbstractQuery implements SpiQuery<T> {
 
   private static final String DEFAULT_QUERY_NAME = "default";
 
@@ -113,10 +112,6 @@ public class DefaultOrmQuery<T> implements SpiQuery<T> {
 
   private ProfilingListener profilingListener;
 
-  private boolean cancelled;
-
-  private CancelableQuery cancelableQuery;
-
   private Type type;
 
   private String label;
@@ -856,6 +851,7 @@ public class DefaultOrmQuery<T> implements SpiQuery<T> {
     copy.parentNode = parentNode;
     copy.forUpdate = forUpdate;
     copy.rawSql = rawSql;
+    setCancelableQuery(copy); // required to cancel findId query
     return copy;
   }
 
@@ -2046,16 +2042,6 @@ public class DefaultOrmQuery<T> implements SpiQuery<T> {
     return futureFetchAudit;
   }
 
-  @Override
-  public void setCancelableQuery(CancelableQuery cancelableQuery) {
-    lock.lock();
-    try {
-      this.cancelableQuery = cancelableQuery;
-    } finally {
-      lock.unlock();
-    }
-  }
-
   @Override
   public Query<T> setBaseTable(String baseTable) {
     this.baseTable = baseTable;
@@ -2083,28 +2069,6 @@ public class DefaultOrmQuery<T> implements SpiQuery<T> {
     return rootTableAlias != null ? rootTableAlias : defaultAlias;
   }
 
-  @Override
-  public void cancel() {
-    lock.lock();
-    try {
-      if (!cancelled && cancelableQuery != null) {
-        cancelled = true;
-        cancelableQuery.cancel();
-      }
-    } finally {
-      lock.unlock();
-    }
-  }
-
-  @Override
-  public boolean isCancelled() {
-    lock.lock();
-    try {
-      return cancelled;
-    } finally {
-      lock.unlock();
-    }
-  }
 
   @Override
   public Set<String> validate() {

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

@@ -17,7 +17,7 @@ import java.util.function.Predicate;
 /**
  * Default implementation of SQuery - SQL Query.
  */
-public class DefaultRelationalQuery implements SpiSqlQuery {
+public class DefaultRelationalQuery extends AbstractQuery implements SpiSqlQuery {
 
   private static final long serialVersionUID = -1098305779779591068L;
 

ebean-core/src/test/java/org/tests/model/basic/MyEBasicConfigStartup.java

@@ -59,19 +59,19 @@ public class MyEBasicConfigStartup implements ServerConfigStartup {
     @Override
     public void inserted(Object bean) {
       insertCount.incrementAndGet();
-      System.out.println("-- EBasic inserted " + ((EBasic) bean).getId());
+      // System.out.println("-- EBasic inserted " + ((EBasic) bean).getId());
     }
 
     @Override
     public void updated(Object bean, Set<String> updatedProperties) {
       updateCount.incrementAndGet();
-      System.out.println("-- EBasic updated " + ((EBasic) bean).getId() + " updatedProperties: " + updatedProperties);
+      // System.out.println("-- EBasic updated " + ((EBasic) bean).getId() + " updatedProperties: " + updatedProperties);
     }
 
     @Override
     public void deleted(Object bean) {
       deleteCount.incrementAndGet();
-      System.out.println("-- EBasic deleted " + ((EBasic) bean).getId());
+      // System.out.println("-- EBasic deleted " + ((EBasic) bean).getId());
     }
 
   }

ebean-core/src/test/java/org/tests/query/cancel/EBasicDto.java

@@ -0,0 +1,28 @@
+package org.tests.query.cancel;
+
+import org.tests.model.basic.EBasic.Status;
+
+/**
+ * DTO for Ebasic Queries.
+ */
+public class EBasicDto {
+  private Integer id;
+
+  private Status status;
+
+  public Integer getId() {
+    return id;
+  }
+
+  public void setId(Integer id) {
+    this.id = id;
+  }
+
+  public Status getStatus() {
+    return status;
+  }
+
+  public void setStatus(Status status) {
+    this.status = status;
+  }
+}

ebean-core/src/test/java/org/tests/query/cancel/SlowDownEBasic.java

@@ -0,0 +1,57 @@
+package org.tests.query.cancel;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.sql.Statement;
+
+import org.h2.api.Trigger;
+
+import io.ebean.DB;
+import io.ebean.Transaction;
+
+/**
+ * Class to artificially slow down selects on 'e_basic' table
+ */
+public class SlowDownEBasic implements Trigger {
+
+  private static int wait;
+
+  private static boolean triggerInstalled;
+
+  @Override
+  public void init(final Connection conn, final String schemaName, final String triggerName, final String tableName,
+      final boolean before, final int type) {
+  }
+
+  @Override
+  public void fire(final Connection conn, final Object[] oldRow, final Object[] newRow) {
+    try {
+      Thread.sleep(wait);
+    } catch (InterruptedException e) {
+      // nop
+    }
+  }
+
+  @Override
+  public void close() {
+  }
+
+  @Override
+  public void remove() {
+  }
+
+
+  public static void setSelectWaitMillis(final int wait) throws SQLException {
+    SlowDownEBasic.wait = wait;
+    if (triggerInstalled) {
+      return;
+    }
+    triggerInstalled = true;
+    try (Transaction txn = DB.beginTransaction(); Statement stmt = txn.getConnection().createStatement()) {
+
+      stmt.execute("CREATE TRIGGER SLOW_DOWN_E_BASIC BEFORE SELECT ON e_basic " + "CALL \""
+          + SlowDownEBasic.class.getName() + "\"");
+      txn.commit();
+    }
+  }
+}

ebean-core/src/test/java/org/tests/query/cancel/SqlQueryCancelTest.java

@@ -0,0 +1,362 @@
+package org.tests.query.cancel;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.sql.SQLException;
+import java.util.UUID;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+import javax.persistence.PersistenceException;
+
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.tests.model.basic.EBasic;
+
+import io.ebean.BaseTestCase;
+import io.ebean.DB;
+import io.ebean.DtoQuery;
+import io.ebean.Query;
+import io.ebean.QueryIterator;
+import io.ebean.SqlQuery;
+import io.ebean.annotation.ForPlatform;
+import io.ebean.annotation.Platform;
+
+/**
+ * Tests, if all kind of queries are cancelable. There are two ways how to
+ * cancel a query: <br/>
+ * <b>At begin:</b>
+ * 
+ * <pre>
+ * query = DB.find(...)
+ * query.cancel();
+ * query.findList();
+ * </pre>
+ * 
+ * The query was caneled before executing. In this case we do hit the DB driver
+ * <br/>
+ * <br/>
+ * <b>During run:</b>
+ * 
+ * <pre>
+ * // Thread 1:              Thread 2
+ * query = DB.find(...)
+ * query.findList();
+ *    ...finding
+ *     ...finding            query.cancel();
+ *      ...JDBC-Exception
+ * </pre>
+ * 
+ * The test tries to simulate a slow query by installing the
+ * {@link SlowDownEBasic} 'SELECT' trigger. The trigger can be configured to
+ * wait 3 * <code>timing</code> ms and a second thread will cancel the query in
+ * <code>timing</code> ms.
+ * 
+ * in this case, we expect a JDBC exception from the driver. <br/>
+ * <br/>
+ * NOTE:<br/>
+ * H2 checks the cancel flag in org.h2.command.Prepared::setCurrentRowNumber
+ * only every 128th row. So we need at least 128 models and we cannot check
+ * queries like findCount or findOne, because they only return one row.
+ * 
+ * @author Roland Praml, FOCONIS AG
+ *
+ */
+public class SqlQueryCancelTest extends BaseTestCase {
+
+  private int timing = 10;
+  
+  @BeforeClass
+  public static void setupTestData() throws SQLException {
+    for (int i = 0; i < 128; i++) {
+      EBasic model = new EBasic("Basic " + i);
+      DB.save(model);
+    }
+    SlowDownEBasic.setSelectWaitMillis(0);
+  }
+
+  @Test
+  public void cancelSqlQueryAtBegin() throws SQLException {
+    doCancelSqlAtBegin(SqlQuery::findList);
+    doCancelSqlAtBegin(SqlQuery::findOne);
+    doCancelSqlAtBegin(q -> q.findEach(e -> {}));
+    doCancelSqlAtBegin(q -> q.findEachWhile(e -> true));
+  }
+
+  @ForPlatform(Platform.H2)
+  @Test
+  public void cancelSqlDuringRun() throws SQLException {
+
+    doCancelSqlDuringRun(SqlQuery::findList);
+    // doCancelSqlDuringRun(q -> q.setMaxRows(1).findOne());
+    // findOne cannot be tested, as H2 does the cancel check every 128 rows only
+    doCancelSqlDuringRun(q -> q.findEach(e -> {}));
+    doCancelSqlDuringRun(q -> q.findEachWhile(e -> true));
+  }
+
+  
+  @Test
+  public void cancelOrmQueryAtBegin() throws SQLException {
+    doCancelOrmAtBegin(Query::findCount); 
+    doCancelOrmAtBegin(Query::findFutureCount);
+    // We cannot test 'findCount' due H2 restrictions
+    doCancelOrmAtBegin(Query::findFutureIds);
+    doCancelOrmAtBegin(Query::findFutureList);
+    doCancelOrmAtBegin(Query::findIds);
+    doCancelOrmAtBegin(Query::findIterate);
+    doCancelOrmAtBegin(Query::findList);
+    doCancelOrmAtBegin(Query::findMap);
+    doCancelOrmAtBegin(Query::findOne);
+    doCancelOrmAtBegin(q -> q.setMaxRows(1000).findPagedList().getList()); // untested
+    doCancelOrmAtBegin(Query::findSet);
+    doCancelOrmAtBegin(Query::findSingleAttribute);
+    doCancelOrmAtBegin(Query::findSingleAttributeList);
+    doCancelOrmAtBegin(Query::findStream);
+   //  testDuringRun(Query::findVersions);
+    // EBasic has no history support, but it should work if @History is added
+    doCancelOrmAtBegin(q -> q.findEach(e -> {}));
+    doCancelOrmAtBegin(q -> q.findEachWhile(e -> true));
+  }
+
+  @ForPlatform(Platform.H2)
+  @Test
+  public void cancelOrmDuringRun() throws Throwable {
+    // doCancelOrmDuringRun(Query::findCount);
+    // testDuringRunFuture(Query::findFutureCount);
+    // We cannot test 'findCount' due H2 restrictions
+    doCancelOrmFutureDuringRun(Query::findFutureIds);
+    doCancelOrmFutureDuringRun(Query::findFutureList);
+    doCancelOrmDuringRun(Query::findIds);
+    doCancelOrmDuringRun(Query::findIterate);
+    doCancelOrmDuringRun(Query::findList);
+    doCancelOrmDuringRun(Query::findMap);
+    // doCancelOrmDuringRun(q -> q.setMaxRows(1).findOne());
+    // findOne cannot be tested, as H2 does the cancel check every 128 rows only
+    doCancelOrmDuringRun(q -> q.setMaxRows(1000).findPagedList().getList()); // untested
+    doCancelOrmDuringRun(Query::findSet);
+    doCancelOrmDuringRun(Query::findSingleAttribute);
+    doCancelOrmDuringRun(Query::findSingleAttributeList);
+    doCancelOrmDuringRun(Query::findStream);
+   //  testDuringRun(Query::findVersions);
+    // EBasic has no history support, but it should work if @History is added
+    doCancelOrmDuringRun(q -> q.findEach(e -> {}));
+    doCancelOrmDuringRun(q -> q.findEachWhile(e -> true));
+  }
+
+  @Test
+  public void cancelOrmDuringIterate() throws SQLException {
+
+    Query<EBasic> query = DB.find(EBasic.class);
+
+    QueryIterator<EBasic> iter = query.findIterate();
+    assertThat(iter.hasNext()).isTrue();
+    query.cancel();
+    assertThat(iter.next()).isNotNull();
+
+    // We might have 100 entities in a buffer. So we must iterate through all.
+    assertThatThrownBy(() -> {
+      while(iter.hasNext()) iter.next();
+    })
+      .isInstanceOf(PersistenceException.class)
+      .hasMessageContaining("Query was cancelled");
+  }
+
+  @Test
+  public void cancelOrmDtoQueryAtBegin() throws SQLException {
+
+    doCancelOrmDtoAtBegin(DtoQuery::findIterate);
+    doCancelOrmDtoAtBegin(DtoQuery::findList);
+    doCancelOrmDtoAtBegin(DtoQuery::findOne);
+    doCancelOrmDtoAtBegin(DtoQuery::findStream);
+    doCancelOrmDtoAtBegin(q -> q.findEach(e -> {}));
+    doCancelOrmDtoAtBegin(q -> q.findEachWhile(e -> true));
+  }
+
+  @ForPlatform(Platform.H2)
+  @Test
+  public void cancelOrmDtoDuringRun() throws SQLException {
+
+    doCancelOrmDtoDuringRun(DtoQuery::findIterate);
+    doCancelOrmDtoDuringRun(DtoQuery::findList);
+    // doCancelOrmDtoDuringRun(q -> q.setMaxRows(1).findOne());
+    // findOne cannot be tested, as H2 does the cancel check every 128 rows only
+    doCancelOrmDtoDuringRun(DtoQuery::findStream);
+    doCancelOrmDtoDuringRun(q -> q.findEach(e -> {}));
+    doCancelOrmDtoDuringRun(q -> q.findEachWhile(e -> true));
+  }
+
+  @Test
+  public void cancelOrmDtoDuringIterate() throws SQLException {
+
+    DtoQuery<EBasicDto> query = DB.find(EBasic.class).select("id,status").asDto(EBasicDto.class);
+
+    QueryIterator<EBasicDto> iter = query.findIterate();
+    assertThat(iter.hasNext()).isTrue();
+    query.cancel();
+    assertThat(iter.next()).isNotNull();
+
+    // We might have 100 entities in a buffer. So we must iterate through all.
+    assertThatThrownBy(() -> {
+      while(iter.hasNext()) iter.next();
+    })
+      .isInstanceOf(PersistenceException.class)
+      .hasMessageContaining("Query was cancelled");
+  }
+  
+  @Test
+  public void cancelSqlDtoQueryAtBegin() throws SQLException {
+
+    doCancelSqlDtoAtBegin(DtoQuery::findIterate);
+    doCancelSqlDtoAtBegin(DtoQuery::findList);
+    doCancelSqlDtoAtBegin(DtoQuery::findOne);
+    doCancelSqlDtoAtBegin(DtoQuery::findStream);
+    doCancelSqlDtoAtBegin(q -> q.findEach(e -> {}));
+    doCancelSqlDtoAtBegin(q -> q.findEachWhile(e -> true));
+  }
+
+  @ForPlatform(Platform.H2)
+  @Test
+  public void cancelSqlDtoDuringRun() throws SQLException {
+
+    //doCancelSqlDtoDuringRun(DtoQuery::findIterate);
+    doCancelSqlDtoDuringRun(DtoQuery::findList);
+    // doCancelSqlDtoDuringRun(q -> q.setMaxRows(1).findOne());
+    // findOne cannot be tested, as H2 does the cancel check every 128 rows only
+    doCancelSqlDtoDuringRun(DtoQuery::findStream);
+    doCancelSqlDtoDuringRun(q -> q.findEach(e -> {}));
+    doCancelSqlDtoDuringRun(q -> q.findEachWhile(e -> true));
+  }
+
+  @Test
+  public void cancelSqlDtoDuringIterate() throws SQLException {
+
+    DtoQuery<EBasicDto> query = DB.findDto(EBasicDto.class, "select id, status from e_basic");
+
+    QueryIterator<EBasicDto> iter = query.findIterate();
+    assertThat(iter.hasNext()).isTrue();
+    query.cancel();
+    assertThat(iter.next()).isNotNull();
+
+    // We might have 100 entities in a buffer. So we must iterate through all.
+    assertThatThrownBy(() -> {
+      while(iter.hasNext()) iter.next();
+    })
+      .isInstanceOf(PersistenceException.class)
+      .hasMessageContaining("Query was cancelled");
+  }
+
+  private void doCancelSqlAtBegin(Consumer<SqlQuery> test) throws SQLException {
+    SqlQuery query = DB.sqlQuery("select * from e_basic");
+    query.cancel();
+    assertThatThrownBy(() -> test.accept(query))
+        .isInstanceOf(PersistenceException.class)
+        .hasMessageContaining("Query was cancelled");
+  }
+
+  private void doCancelSqlDuringRun(Consumer<SqlQuery> test) throws SQLException {
+    SqlQuery warmup = DB.sqlQuery("select * from e_basic");
+    test.accept(warmup);
+    SqlQuery query = DB.sqlQuery("select * from e_basic");
+    executeDelayed(query::cancel);
+    assertThatThrownBy(() -> test.accept(query))
+        .isInstanceOf(PersistenceException.class)
+        .hasCauseInstanceOf(org.h2.jdbc.JdbcSQLTimeoutException.class);
+  }
+
+  private void doCancelOrmAtBegin(Consumer<Query<EBasic>> test) throws SQLException {
+    Query<EBasic> query = DB.find(EBasic.class);
+    query.cancel();
+    assertThatThrownBy(() -> test.accept(query))
+        .isInstanceOf(PersistenceException.class)
+        .hasMessageContaining("Query was cancelled");
+  }
+
+  private void doCancelOrmDuringRun(Consumer<Query<EBasic>> test) throws SQLException {
+    Query<EBasic> warmup = DB.find(EBasic.class);
+    test.accept(warmup);
+    Query<EBasic> warmup2 = DB.find(EBasic.class);
+    test.accept(warmup2);
+    Query<EBasic> query = DB.find(EBasic.class);
+    executeDelayed(query::cancel);
+    assertThatThrownBy(() -> test.accept(query))
+        .isInstanceOf(PersistenceException.class)
+        .hasCauseInstanceOf(org.h2.jdbc.JdbcSQLTimeoutException.class);
+  }
+
+  private void doCancelOrmFutureDuringRun(Function<Query<EBasic>, Future<?>> test) throws SQLException, InterruptedException, ExecutionException {
+    Query<EBasic> warmup = DB.find(EBasic.class);
+    test.apply(warmup).get();
+    
+    Query<EBasic> query = DB.find(EBasic.class);
+    executeDelayed(query::cancel);
+    assertThatThrownBy(() -> {
+      try {
+        test.apply(query).get();
+      } catch (ExecutionException ee) {
+        throw ee.getCause();
+      }
+    })
+      .isInstanceOf(PersistenceException.class)
+      .hasCauseInstanceOf(org.h2.jdbc.JdbcSQLTimeoutException.class);
+  }
+
+  private void doCancelOrmDtoAtBegin(Consumer<DtoQuery<EBasicDto>> test) throws SQLException {
+    DtoQuery<EBasicDto> query = DB.find(EBasic.class).select("id,status").asDto(EBasicDto.class);
+    query.cancel();
+    assertThatThrownBy(() -> test.accept(query))
+        .isInstanceOf(PersistenceException.class)
+        .hasMessageContaining("Query was cancelled");
+  }
+  
+  private void doCancelOrmDtoDuringRun(Consumer<DtoQuery<EBasicDto>> test) throws SQLException {
+    DtoQuery<EBasicDto> warmup = DB.find(EBasic.class).select("id,status").asDto(EBasicDto.class);
+    test.accept(warmup);
+    
+    DtoQuery<EBasicDto> query = DB.find(EBasic.class).select("id,status").asDto(EBasicDto.class);
+    executeDelayed(query::cancel);
+    assertThatThrownBy(() -> test.accept(query))
+        .isInstanceOf(PersistenceException.class)
+        .hasCauseInstanceOf(org.h2.jdbc.JdbcSQLTimeoutException.class);
+  }
+  
+  private void doCancelSqlDtoAtBegin(Consumer<DtoQuery<EBasicDto>> test) throws SQLException {
+    DtoQuery<EBasicDto> query = DB.findDto(EBasicDto.class, "select id, status from e_basic");
+    query.cancel();
+    assertThatThrownBy(() -> test.accept(query))
+        .isInstanceOf(PersistenceException.class)
+        .hasMessageContaining("Query was cancelled");
+  }
+
+  private void doCancelSqlDtoDuringRun(Consumer<DtoQuery<EBasicDto>> test) throws SQLException {
+    DtoQuery<EBasicDto> warmup = DB.findDto(EBasicDto.class, "select id, status from e_basic");
+    test.accept(warmup);
+    
+    DtoQuery<EBasicDto> query = DB.findDto(EBasicDto.class, "select id, status from e_basic");
+    executeDelayed(query::cancel);
+    assertThatThrownBy(() -> test.accept(query))
+        .isInstanceOf(PersistenceException.class)
+        .hasCauseInstanceOf(org.h2.jdbc.JdbcSQLTimeoutException.class);
+  }
+  
+  private void executeDelayed(Runnable r) throws SQLException {
+    // We modify the DB here. Otherwise we may hit an internal H2 cache, if the
+    // same query is performed. Queries from the cache cannot be canceled.
+    EBasic makeDbDirty = new EBasic("Basic " + UUID.randomUUID());
+    DB.save(makeDbDirty);
+    SlowDownEBasic.setSelectWaitMillis(timing * 3);
+    new Thread(() -> {
+      try {
+        Thread.sleep(timing);
+        r.run();
+        SlowDownEBasic.setSelectWaitMillis(0);
+      } catch (Exception e) {
+        e.printStackTrace();
+      }
+    }).start();
+  }
+
+}