asterixdb-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From buyin...@apache.org
Subject [2/2] asterixdb git commit: Add a REST endpoint for query cancellation.
Date Mon, 13 Mar 2017 16:08:05 GMT
Add a REST endpoint for query cancellation.

This change also includes the following parts:
- Fix failure handling and potentially thread leakage in
  MaterializingPipelinedPartition;
- Fix failure handling in PartitionDataWriter;
- Add a new test suite: SqlppExecutionWithCancellationTest.

Change-Id: I2936ac83f71bbef533e2695ed0a2b220c23fc483
Reviewed-on: https://asterix-gerrit.ics.uci.edu/1564
Tested-by: Jenkins <jenkins@fulliautomatix.ics.uci.edu>
Integration-Tests: Jenkins <jenkins@fulliautomatix.ics.uci.edu>
Reviewed-by: Till Westmann <tillw@apache.org>


Project: http://git-wip-us.apache.org/repos/asf/asterixdb/repo
Commit: http://git-wip-us.apache.org/repos/asf/asterixdb/commit/343cd457
Tree: http://git-wip-us.apache.org/repos/asf/asterixdb/tree/343cd457
Diff: http://git-wip-us.apache.org/repos/asf/asterixdb/diff/343cd457

Branch: refs/heads/master
Commit: 343cd457d3c12e584729b682684ba87c9886ddd8
Parents: 61401f9
Author: Yingyi Bu <yingyi@couchbase.com>
Authored: Mon Mar 13 00:27:56 2017 -0700
Committer: Yingyi Bu <buyingyi@gmail.com>
Committed: Mon Mar 13 09:07:38 2017 -0700

----------------------------------------------------------------------
 .../asterix/translator/IStatementExecutor.java  |  16 ++-
 .../translator/IStatementExecutorContext.java   |  56 ++++++++
 .../api/http/ctx/StatementExecutorContext.java  |  46 ++++++
 .../asterix/api/http/server/ApiServlet.java     |   3 +-
 .../api/http/server/DiagnosticsApiServlet.java  |   2 +-
 .../http/server/QueryCancellationServlet.java   |  82 +++++++++++
 .../api/http/server/QueryServiceServlet.java    |  10 +-
 .../asterix/api/http/server/RestApiServlet.java |   2 +-
 .../api/http/servlet/ServletConstants.java      |   3 +-
 .../asterix/api/java/AsterixJavaClient.java     |   3 +-
 .../asterix/app/translator/QueryTranslator.java |  58 ++++----
 .../hyracks/bootstrap/CCApplication.java        |   7 +-
 .../servlet/QueryCancellationServletTest.java   |  92 ++++++++++++
 .../test/common/CancellationTestExecutor.java   |  91 ++++++++++++
 .../asterix/test/common/ResultExtractor.java    |   5 +-
 .../asterix/test/common/TestExecutor.java       |   8 +-
 .../asterix/test/runtime/APIExecutionTest.java  |   3 +-
 .../runtime/AqlExecutionFullParallelismIT.java  |   3 +-
 .../runtime/AqlExecutionLessParallelismIT.java  |   3 +-
 .../asterix/test/runtime/AqlExecutionTest.java  |   3 +-
 .../ClusterStateDefaultParameterTest.java       |   3 +-
 ...lusterStateExecutionFullParallelismTest.java |   3 +-
 ...lusterStateExecutionLessParallelismTest.java |   3 +-
 .../asterix/test/runtime/LangExecutionUtil.java |  63 ++++++--
 .../SqlppExecutionFullParallelismIT.java        |   3 +-
 .../SqlppExecutionLessParallelismIT.java        |   3 +-
 .../test/runtime/SqlppExecutionTest.java        |   3 +-
 .../SqlppExecutionWithCancellationTest.java     | 100 +++++++++++++
 .../types/any-object/any-object.2.query.aql     |   1 +
 .../types/any-object/any-object.2.query.sqlpp   |   4 +-
 .../results/types/any-object/any-object.2.adm   |  72 +---------
 .../org/apache/asterix/common/context/Info.java |   4 -
 .../apache/asterix/common/utils/Servlets.java   |   1 +
 .../org/apache/hyracks/control/nc/Task.java     |   2 +-
 .../hyracks/control/nc/io/FileHandle.java       |  11 +-
 .../apache/hyracks/control/nc/io/IOManager.java |   2 +-
 .../MaterializingPipelinedPartition.java        | 142 ++++++++++++-------
 .../nc/partitions/PipelinedPartition.java       |   4 +-
 .../control/nc/work/CleanupJobletWork.java      |   9 +-
 .../dataflow/common/io/RunFileReader.java       |   4 +
 .../AbstractReplicateOperatorDescriptor.java    |   2 +-
 .../std/connectors/PartitionDataWriter.java     |  44 ++++--
 .../std/misc/MaterializerTaskState.java         |  19 ++-
 .../misc/MaterializingOperatorDescriptor.java   |   6 +-
 .../std/sort/AbstractExternalSortRunMerger.java |  19 ++-
 .../TreeIndexStatsOperatorNodePushable.java     |   7 +-
 46 files changed, 797 insertions(+), 233 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/asterixdb/blob/343cd457/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/IStatementExecutor.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/IStatementExecutor.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/IStatementExecutor.java
index 2066f73..92c487b 100644
--- a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/IStatementExecutor.java
+++ b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/IStatementExecutor.java
@@ -77,7 +77,7 @@ public interface IStatementExecutor {
     }
 
     /**
-     * Compiles and execute a list of statements.
+     * Compiles and execute a list of statements, without passing in client context id and context.
      *
      * @param hcc
      *            A Hyracks client connection that is used to submit a jobspec to Hyracks.
@@ -85,13 +85,15 @@ public interface IStatementExecutor {
      *            A Hyracks dataset client object that is used to read the results.
      * @param resultDelivery
      *            The {@code ResultDelivery} kind required for queries in the list of statements
+     * @param stats
+     *            a reference to write the stats of executed queries
      * @throws Exception
      */
-    void compileAndExecute(IHyracksClientConnection hcc, IHyracksDataset hdc, ResultDelivery resultDelivery)
-            throws Exception;
+    void compileAndExecute(IHyracksClientConnection hcc, IHyracksDataset hdc, ResultDelivery resultDelivery,
+            Stats stats) throws Exception;
 
     /**
-     * Compiles and execute a list of statements.
+     * Compiles and execute a list of statements, with passing in client context id and context.
      *
      * @param hcc
      *            A Hyracks client connection that is used to submit a jobspec to Hyracks.
@@ -101,10 +103,14 @@ public interface IStatementExecutor {
      *            The {@code ResultDelivery} kind required for queries in the list of statements
      * @param stats
      *            a reference to write the stats of executed queries
+     * @param clientContextId
+     *            the client context id for the query
+     * @param ctx
+     *            the context that contains the meta information for all queries
      * @throws Exception
      */
     void compileAndExecute(IHyracksClientConnection hcc, IHyracksDataset hdc, ResultDelivery resultDelivery,
-            Stats stats) throws Exception;
+            Stats stats, String clientContextId, IStatementExecutorContext ctx) throws Exception;
 
     /**
      * rewrites and compiles query into a hyracks job specifications

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/343cd457/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/IStatementExecutorContext.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/IStatementExecutorContext.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/IStatementExecutorContext.java
new file mode 100644
index 0000000..81e1ebf
--- /dev/null
+++ b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/IStatementExecutorContext.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.asterix.translator;
+
+import org.apache.hyracks.api.job.JobId;
+
+/**
+ * The context for statement executors, which maintains the meta information of all queries.
+ * TODO(yingyi): also maintain the mapping from server generated request ids to jobs.
+ */
+public interface IStatementExecutorContext {
+
+    /**
+     * Gets the Hyracks JobId from the user-provided client context id.
+     *
+     * @param clientContextId,
+     *            a user provided client context id.
+     * @return the Hyracks job id of class {@link org.apache.hyracks.api.job.JobId}.
+     */
+    JobId getJobIdFromClientContextId(String clientContextId);
+
+    /**
+     * Puts a client context id for a statement and the corresponding Hyracks job id.
+     *
+     * @param clientContextId,
+     *            a user provided client context id.
+     * @param jobId,
+     *            the Hyracks job id of class {@link org.apache.hyracks.api.job.JobId}.
+     */
+    void put(String clientContextId, JobId jobId);
+
+    /**
+     * Removes the information about the query corresponding to a user-provided client context id.
+     *
+     * @param clientContextId,
+     *            a user provided client context id.
+     */
+    JobId removeJobIdFromClientContextId(String clientContextId);
+}

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/343cd457/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/ctx/StatementExecutorContext.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/ctx/StatementExecutorContext.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/ctx/StatementExecutorContext.java
new file mode 100644
index 0000000..7c06762
--- /dev/null
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/ctx/StatementExecutorContext.java
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.asterix.api.http.ctx;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.asterix.translator.IStatementExecutorContext;
+import org.apache.hyracks.api.job.JobId;
+
+public class StatementExecutorContext implements IStatementExecutorContext {
+
+    private final Map<String, JobId> runningQueries = new ConcurrentHashMap<>();
+
+    @Override
+    public JobId getJobIdFromClientContextId(String clientContextId) {
+        return runningQueries.get(clientContextId);
+    }
+
+    @Override
+    public void put(String clientContextId, JobId jobId) {
+        runningQueries.put(clientContextId, jobId);
+    }
+
+    @Override
+    public JobId removeJobIdFromClientContextId(String clientContextId) {
+        return runningQueries.remove(clientContextId);
+    }
+}

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/343cd457/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/ApiServlet.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/ApiServlet.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/ApiServlet.java
index 7a33c0e..d91d5fc 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/ApiServlet.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/ApiServlet.java
@@ -145,7 +145,8 @@ public class ApiServlet extends AbstractServlet {
                     compilationProvider, componentProvider);
             double duration;
             long startTime = System.currentTimeMillis();
-            translator.compileAndExecute(hcc, hds, IStatementExecutor.ResultDelivery.IMMEDIATE);
+            translator.compileAndExecute(hcc, hds, IStatementExecutor.ResultDelivery.IMMEDIATE,
+                    new IStatementExecutor.Stats());
             long endTime = System.currentTimeMillis();
             duration = (endTime - startTime) / 1000.00;
             out.println(HTML_STATEMENT_SEPARATOR);

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/343cd457/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/DiagnosticsApiServlet.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/DiagnosticsApiServlet.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/DiagnosticsApiServlet.java
index de227eb..788927f 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/DiagnosticsApiServlet.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/DiagnosticsApiServlet.java
@@ -83,7 +83,7 @@ public class DiagnosticsApiServlet extends NodeControllerDetailsApiServlet {
     private ObjectNode getClusterDiagnosticsJSON() throws Exception {
         ObjectMapper om = new ObjectMapper();
         IHyracksClientConnection hcc = (IHyracksClientConnection) ctx.get(HYRACKS_CONNECTION_ATTR);
-        ExecutorService executor = (ExecutorService) ctx.get(ServletConstants.EXECUTOR_SERVICE);
+        ExecutorService executor = (ExecutorService) ctx.get(ServletConstants.EXECUTOR_SERVICE_ATTR);
         Map<String, Future<ObjectNode>> ccFutureData = new HashMap<>();
         ccFutureData.put("threaddump",
                 executor.submit(() -> fixupKeys((ObjectNode) om.readTree(hcc.getThreadDump(null)))));

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/343cd457/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/QueryCancellationServlet.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/QueryCancellationServlet.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/QueryCancellationServlet.java
new file mode 100644
index 0000000..4405d29
--- /dev/null
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/QueryCancellationServlet.java
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.asterix.api.http.server;
+
+import java.io.IOException;
+import java.util.concurrent.ConcurrentMap;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import org.apache.asterix.api.http.servlet.ServletConstants;
+import org.apache.asterix.translator.IStatementExecutorContext;
+import org.apache.hyracks.api.client.IHyracksClientConnection;
+import org.apache.hyracks.api.job.JobId;
+import org.apache.hyracks.http.api.IServletRequest;
+import org.apache.hyracks.http.api.IServletResponse;
+import org.apache.hyracks.http.server.AbstractServlet;
+
+import io.netty.handler.codec.http.HttpResponseStatus;
+
+/**
+ * The servlet provides a REST API for cancelling an on-going query.
+ */
+public class QueryCancellationServlet extends AbstractServlet {
+    private static final Logger LOGGER = Logger.getLogger(QueryCancellationServlet.class.getName());
+    private static final String CLIENT_CONTEXT_ID = "client_context_id";
+
+    public QueryCancellationServlet(ConcurrentMap<String, Object> ctx, String... paths) {
+        super(ctx, paths);
+    }
+
+    @Override
+    protected void delete(IServletRequest request, IServletResponse response) throws IOException {
+        // gets the parameter client_context_id from the request.
+        String clientContextId = request.getParameter(CLIENT_CONTEXT_ID);
+        if (clientContextId == null) {
+            response.setStatus(HttpResponseStatus.BAD_REQUEST);
+            return;
+        }
+
+        // Retrieves the corresponding Hyracks job id.
+        IStatementExecutorContext runningQueries = (IStatementExecutorContext) ctx
+                .get(ServletConstants.RUNNING_QUERIES_ATTR);
+        IHyracksClientConnection hcc = (IHyracksClientConnection) ctx.get(ServletConstants.HYRACKS_CONNECTION_ATTR);
+        JobId jobId = runningQueries.getJobIdFromClientContextId(clientContextId);
+
+        if (jobId == null) {
+            // response: NOT FOUND
+            response.setStatus(HttpResponseStatus.NOT_FOUND);
+            return;
+        }
+        try {
+            // Cancels the on-going job.
+            hcc.cancelJob(jobId);
+            // Removes the cancelled query from the map activeQueries.
+            runningQueries.removeJobIdFromClientContextId(clientContextId);
+            // response: OK
+            response.setStatus(HttpResponseStatus.OK);
+        } catch (Exception e) {
+            if (LOGGER.isLoggable(Level.WARNING)) {
+                LOGGER.log(Level.WARNING, e.getMessage(), e);
+            }
+            // response: INTERNAL SERVER ERROR
+            response.setStatus(HttpResponseStatus.INTERNAL_SERVER_ERROR);
+        }
+    }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/343cd457/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/QueryServiceServlet.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/QueryServiceServlet.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/QueryServiceServlet.java
index 9d22452..42bb4f9 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/QueryServiceServlet.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/QueryServiceServlet.java
@@ -29,6 +29,8 @@ import java.util.concurrent.ConcurrentMap;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
+import org.apache.asterix.api.http.ctx.StatementExecutorContext;
+import org.apache.asterix.api.http.servlet.ServletConstants;
 import org.apache.asterix.app.result.ResultUtil;
 import org.apache.asterix.common.api.IClusterManagementWork;
 import org.apache.asterix.common.config.GlobalConfig;
@@ -43,6 +45,7 @@ import org.apache.asterix.metadata.MetadataManager;
 import org.apache.asterix.runtime.utils.ClusterStateManager;
 import org.apache.asterix.translator.IStatementExecutor;
 import org.apache.asterix.translator.IStatementExecutor.Stats;
+import org.apache.asterix.translator.IStatementExecutorContext;
 import org.apache.asterix.translator.IStatementExecutorFactory;
 import org.apache.asterix.translator.SessionConfig;
 import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
@@ -65,6 +68,7 @@ public class QueryServiceServlet extends AbstractQueryApiServlet {
     private final ILangCompilationProvider compilationProvider;
     private final IStatementExecutorFactory statementExecutorFactory;
     private final IStorageComponentProvider componentProvider;
+    private final IStatementExecutorContext queryCtx = new StatementExecutorContext();
 
     public QueryServiceServlet(ConcurrentMap<String, Object> ctx, String[] paths,
             ILangCompilationProvider compilationProvider, IStatementExecutorFactory statementExecutorFactory,
@@ -73,6 +77,7 @@ public class QueryServiceServlet extends AbstractQueryApiServlet {
         this.compilationProvider = compilationProvider;
         this.statementExecutorFactory = statementExecutorFactory;
         this.componentProvider = componentProvider;
+        ctx.put(ServletConstants.RUNNING_QUERIES_ATTR, queryCtx);
     }
 
     @Override
@@ -425,7 +430,8 @@ public class QueryServiceServlet extends AbstractQueryApiServlet {
             IStatementExecutor translator =
                     statementExecutorFactory.create(statements, sessionConfig, compilationProvider, componentProvider);
             execStart = System.nanoTime();
-            translator.compileAndExecute(getHyracksClientConnection(), getHyracksDataset(), delivery, stats);
+            translator.compileAndExecute(getHyracksClientConnection(), getHyracksDataset(), delivery, stats,
+                    param.clientContextID, queryCtx);
             execEnd = System.nanoTime();
             printStatus(resultWriter, ResultDelivery.ASYNC == delivery ? ResultStatus.RUNNING : ResultStatus.SUCCESS);
         } catch (AsterixException | TokenMgrError | org.apache.asterix.aqlplus.parser.TokenMgrError pe) {
@@ -459,4 +465,4 @@ public class QueryServiceServlet extends AbstractQueryApiServlet {
             LOGGER.warning("Error flushing output writer");
         }
     }
-}
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/343cd457/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/RestApiServlet.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/RestApiServlet.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/RestApiServlet.java
index bdc9d62..d7edb23 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/RestApiServlet.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/RestApiServlet.java
@@ -203,7 +203,7 @@ public abstract class RestApiServlet extends AbstractServlet {
             MetadataManager.INSTANCE.init();
             IStatementExecutor translator = statementExecutorFactory.create(aqlStatements, sessionConfig,
                     compilationProvider, componentProvider);
-            translator.compileAndExecute(hcc, hds, resultDelivery);
+            translator.compileAndExecute(hcc, hds, resultDelivery, new IStatementExecutor.Stats());
         } catch (AsterixException | TokenMgrError | org.apache.asterix.aqlplus.parser.TokenMgrError pe) {
             response.setStatus(HttpResponseStatus.INTERNAL_SERVER_ERROR);
             GlobalConfig.ASTERIX_LOGGER.log(Level.SEVERE, pe.getMessage(), pe);

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/343cd457/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/servlet/ServletConstants.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/servlet/ServletConstants.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/servlet/ServletConstants.java
index 5b96cab..a9d4e22 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/servlet/ServletConstants.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/servlet/ServletConstants.java
@@ -22,7 +22,8 @@ public class ServletConstants {
     public static final String HYRACKS_CONNECTION_ATTR = "org.apache.asterix.HYRACKS_CONNECTION";
     public static final String HYRACKS_DATASET_ATTR = "org.apache.asterix.HYRACKS_DATASET";
     public static final String ASTERIX_APP_CONTEXT_INFO_ATTR = "org.apache.asterix.APP_CONTEXT_INFO";
-    public static final String EXECUTOR_SERVICE = "org.apache.asterix.EXECUTOR_SERVICE";
+    public static final String EXECUTOR_SERVICE_ATTR = "org.apache.asterix.EXECUTOR_SERVICE_ATTR";
+    public static final String RUNNING_QUERIES_ATTR = "org.apache.asterix.RUNINNG_QUERIES";
 
     private ServletConstants() {
     }

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/343cd457/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/java/AsterixJavaClient.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/java/AsterixJavaClient.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/java/AsterixJavaClient.java
index c09f8cb..d03e574 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/java/AsterixJavaClient.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/java/AsterixJavaClient.java
@@ -104,7 +104,8 @@ public class AsterixJavaClient {
 
         IStatementExecutor translator =
                 statementExecutorFactory.create(statements, conf, compilationProvider, storageComponentProvider);
-        translator.compileAndExecute(hcc, null, QueryTranslator.ResultDelivery.IMMEDIATE);
+        translator.compileAndExecute(hcc, null, QueryTranslator.ResultDelivery.IMMEDIATE,
+                new IStatementExecutor.Stats());
         writer.flush();
     }
 

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/343cd457/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
index 1e4d866..26a6ebd 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
@@ -156,6 +156,7 @@ import org.apache.asterix.translator.CompiledStatements.CompiledLoadFromFileStat
 import org.apache.asterix.translator.CompiledStatements.CompiledUpsertStatement;
 import org.apache.asterix.translator.CompiledStatements.ICompiledDmlStatement;
 import org.apache.asterix.translator.IStatementExecutor;
+import org.apache.asterix.translator.IStatementExecutorContext;
 import org.apache.asterix.translator.SessionConfig;
 import org.apache.asterix.translator.TypeTranslator;
 import org.apache.asterix.translator.util.ValidateUtil;
@@ -232,27 +233,15 @@ public class QueryTranslator extends AbstractLangTranslator implements IStatemen
         return functionDecls;
     }
 
-    /**
-     * Compiles and submits for execution a list of AQL statements.
-     *
-     * @param hcc
-     *            A Hyracks client connection that is used to submit a jobspec to Hyracks.
-     * @param hdc
-     *            A Hyracks dataset client object that is used to read the results.
-     * @param resultDelivery
-     *            True if the results should be read asynchronously or false if we should wait for results to be read.
-     * @return A List<QueryResult> containing a QueryResult instance corresponding to each submitted query.
-     * @throws Exception
-     */
     @Override
-    public void compileAndExecute(IHyracksClientConnection hcc, IHyracksDataset hdc, ResultDelivery resultDelivery)
-            throws Exception {
-        compileAndExecute(hcc, hdc, resultDelivery, new Stats());
+    public void compileAndExecute(IHyracksClientConnection hcc, IHyracksDataset hdc, ResultDelivery resultDelivery,
+            Stats stats) throws Exception {
+        compileAndExecute(hcc, hdc, resultDelivery, stats, null, null);
     }
 
     @Override
     public void compileAndExecute(IHyracksClientConnection hcc, IHyracksDataset hdc, ResultDelivery resultDelivery,
-            Stats stats) throws Exception {
+            Stats stats, String clientContextId, IStatementExecutorContext ctx) throws Exception {
         int resultSetIdCounter = 0;
         FileSplit outputFile = null;
         IAWriterFactory writerFactory = PrinterBasedWriterFactory.INSTANCE;
@@ -329,7 +318,8 @@ public class QueryTranslator extends AbstractLangTranslator implements IStatemen
                             metadataProvider.setResultAsyncMode(resultDelivery == ResultDelivery.ASYNC
                                     || resultDelivery == ResultDelivery.DEFERRED);
                         }
-                        handleInsertUpsertStatement(metadataProvider, stmt, hcc, hdc, resultDelivery, stats, false);
+                        handleInsertUpsertStatement(metadataProvider, stmt, hcc, hdc, resultDelivery, stats, false,
+                                clientContextId, ctx);
                         break;
                     case Statement.Kind.DELETE:
                         handleDeleteStatement(metadataProvider, stmt, hcc, false);
@@ -362,7 +352,8 @@ public class QueryTranslator extends AbstractLangTranslator implements IStatemen
                         metadataProvider.setResultSetId(new ResultSetId(resultSetIdCounter++));
                         metadataProvider.setResultAsyncMode(
                                 resultDelivery == ResultDelivery.ASYNC || resultDelivery == ResultDelivery.DEFERRED);
-                        handleQuery(metadataProvider, (Query) stmt, hcc, hdc, resultDelivery, stats);
+                        handleQuery(metadataProvider, (Query) stmt, hcc, hdc, resultDelivery, stats, clientContextId,
+                                ctx);
                         break;
                     case Statement.Kind.COMPACT:
                         handleCompactStatement(metadataProvider, stmt, hcc);
@@ -1809,8 +1800,8 @@ public class QueryTranslator extends AbstractLangTranslator implements IStatemen
 
     public JobSpecification handleInsertUpsertStatement(MetadataProvider metadataProvider, Statement stmt,
             IHyracksClientConnection hcc, IHyracksDataset hdc, ResultDelivery resultDelivery,
-            IStatementExecutor.Stats stats, boolean compileOnly) throws Exception {
-
+            IStatementExecutor.Stats stats, boolean compileOnly, String clientContextId, IStatementExecutorContext ctx)
+            throws Exception {
         InsertStatement stmtInsertUpsert = (InsertStatement) stmt;
         String dataverseName = getActiveDataverse(stmtInsertUpsert.getDataverseName());
         Query query = stmtInsertUpsert.getQuery();
@@ -1852,7 +1843,7 @@ public class QueryTranslator extends AbstractLangTranslator implements IStatemen
         }
 
         if (stmtInsertUpsert.getReturnExpression() != null) {
-            deliverResult(hcc, hdc, compiler, metadataProvider, locker, resultDelivery, stats);
+            deliverResult(hcc, hdc, compiler, metadataProvider, locker, resultDelivery, stats, clientContextId, ctx);
         } else {
             locker.lock();
             try {
@@ -2371,7 +2362,8 @@ public class QueryTranslator extends AbstractLangTranslator implements IStatemen
     }
 
     protected void handleQuery(MetadataProvider metadataProvider, Query query, IHyracksClientConnection hcc,
-            IHyracksDataset hdc, ResultDelivery resultDelivery, Stats stats) throws Exception {
+            IHyracksDataset hdc, ResultDelivery resultDelivery, Stats stats, String clientContextId,
+            IStatementExecutorContext ctx) throws Exception {
         final IMetadataLocker locker = new IMetadataLocker() {
             @Override
             public void lock() {
@@ -2402,11 +2394,12 @@ public class QueryTranslator extends AbstractLangTranslator implements IStatemen
                 throw e;
             }
         };
-        deliverResult(hcc, hdc, compiler, metadataProvider, locker, resultDelivery, stats);
+        deliverResult(hcc, hdc, compiler, metadataProvider, locker, resultDelivery, stats, clientContextId, ctx);
     }
 
     private void deliverResult(IHyracksClientConnection hcc, IHyracksDataset hdc, IStatementCompiler compiler,
-            MetadataProvider metadataProvider, IMetadataLocker locker, ResultDelivery resultDelivery, Stats stats)
+            MetadataProvider metadataProvider, IMetadataLocker locker, ResultDelivery resultDelivery, Stats stats,
+            String clientContextId, IStatementExecutorContext ctx)
             throws Exception {
         final ResultSetId resultSetId = metadataProvider.getResultSetId();
         switch (resultDelivery) {
@@ -2422,7 +2415,7 @@ public class QueryTranslator extends AbstractLangTranslator implements IStatemen
                                 printed.setTrue();
                                 printed.notify();
                             }
-                        });
+                        }, clientContextId, ctx);
                     } catch (Exception e) {
                         GlobalConfig.ASTERIX_LOGGER.log(Level.SEVERE,
                                 resultDelivery.name() + " job " + "with id " + jobId + " failed", e);
@@ -2439,12 +2432,12 @@ public class QueryTranslator extends AbstractLangTranslator implements IStatemen
                     final ResultReader resultReader = new ResultReader(hdc, id, resultSetId);
                     ResultUtil.printResults(resultReader, sessionConfig, stats,
                             metadataProvider.findOutputRecordType());
-                });
+                }, clientContextId, ctx);
                 break;
             case DEFERRED:
                 createAndRunJob(hcc, compiler, locker, resultDelivery, id -> {
                     ResultUtil.printResultHandle(new ResultHandle(id, resultSetId), sessionConfig);
-                });
+                }, clientContextId, ctx);
                 break;
             default:
                 break;
@@ -2452,7 +2445,8 @@ public class QueryTranslator extends AbstractLangTranslator implements IStatemen
     }
 
     private static JobId createAndRunJob(IHyracksClientConnection hcc, IStatementCompiler compiler,
-            IMetadataLocker locker, ResultDelivery resultDelivery, IResultPrinter printer) throws Exception {
+            IMetadataLocker locker, ResultDelivery resultDelivery, IResultPrinter printer, String clientContextId,
+            IStatementExecutorContext ctx) throws Exception {
         locker.lock();
         try {
             final JobSpecification jobSpec = compiler.compile();
@@ -2460,6 +2454,10 @@ public class QueryTranslator extends AbstractLangTranslator implements IStatemen
                 return JobId.INVALID;
             }
             final JobId jobId = JobUtils.runJob(hcc, jobSpec, false);
+
+            if (ctx != null && clientContextId != null) {
+                ctx.put(clientContextId, jobId); // Adds the running job into the context.
+            }
             if (ResultDelivery.ASYNC == resultDelivery) {
                 printer.print(jobId);
                 hcc.waitForCompletion(jobId);
@@ -2469,6 +2467,10 @@ public class QueryTranslator extends AbstractLangTranslator implements IStatemen
             }
             return jobId;
         } finally {
+            // No matter the job succeeds or fails, removes it into the context.
+            if (ctx != null && clientContextId != null) {
+                ctx.removeJobIdFromClientContextId(clientContextId);
+            }
             locker.unlock();
         }
     }

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/343cd457/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/CCApplication.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/CCApplication.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/CCApplication.java
index c831508..fcc6f1f 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/CCApplication.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/CCApplication.java
@@ -16,6 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
+
 package org.apache.asterix.hyracks.bootstrap;
 
 import static org.apache.asterix.api.http.servlet.ServletConstants.ASTERIX_APP_CONTEXT_INFO_ATTR;
@@ -38,6 +39,7 @@ import org.apache.asterix.api.http.server.FeedServlet;
 import org.apache.asterix.api.http.server.FullApiServlet;
 import org.apache.asterix.api.http.server.NodeControllerDetailsApiServlet;
 import org.apache.asterix.api.http.server.QueryApiServlet;
+import org.apache.asterix.api.http.server.QueryCancellationServlet;
 import org.apache.asterix.api.http.server.QueryResultApiServlet;
 import org.apache.asterix.api.http.server.QueryServiceServlet;
 import org.apache.asterix.api.http.server.QueryStatusApiServlet;
@@ -184,7 +186,7 @@ public class CCApplication extends BaseCCApplication {
         IHyracksClientConnection hcc = getHcc();
         jsonAPIServer.setAttribute(HYRACKS_CONNECTION_ATTR, hcc);
         jsonAPIServer.setAttribute(ASTERIX_APP_CONTEXT_INFO_ATTR, AppContextInfo.INSTANCE);
-        jsonAPIServer.setAttribute(ServletConstants.EXECUTOR_SERVICE,
+        jsonAPIServer.setAttribute(ServletConstants.EXECUTOR_SERVICE_ATTR,
                 ((ClusterControllerService) ccServiceCtx.getControllerService()).getExecutor());
 
         // AQL rest APIs.
@@ -203,6 +205,7 @@ public class CCApplication extends BaseCCApplication {
         addServlet(jsonAPIServer, Servlets.QUERY_STATUS);
         addServlet(jsonAPIServer, Servlets.QUERY_RESULT);
         addServlet(jsonAPIServer, Servlets.QUERY_SERVICE);
+        addServlet(jsonAPIServer, Servlets.RUNNING_REQUESTS);
         addServlet(jsonAPIServer, Servlets.CONNECTOR);
         addServlet(jsonAPIServer, Servlets.SHUTDOWN);
         addServlet(jsonAPIServer, Servlets.VERSION);
@@ -260,6 +263,8 @@ public class CCApplication extends BaseCCApplication {
             case Servlets.SQLPP_DDL:
                 return new DdlApiServlet(ctx, paths, ccExtensionManager.getSqlppCompilationProvider(),
                         getStatementExecutorFactory(), componentProvider);
+            case Servlets.RUNNING_REQUESTS:
+                return new QueryCancellationServlet(ctx, paths);
             case Servlets.QUERY_STATUS:
                 return new QueryStatusApiServlet(ctx, paths);
             case Servlets.QUERY_RESULT:

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/343cd457/asterixdb/asterix-app/src/test/java/org/apache/asterix/api/http/servlet/QueryCancellationServletTest.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/api/http/servlet/QueryCancellationServletTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/api/http/servlet/QueryCancellationServletTest.java
new file mode 100644
index 0000000..5f40a85
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/api/http/servlet/QueryCancellationServletTest.java
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.asterix.api.http.servlet;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.asterix.api.http.ctx.StatementExecutorContext;
+import org.apache.asterix.api.http.server.QueryCancellationServlet;
+import org.apache.asterix.translator.IStatementExecutorContext;
+import org.apache.hyracks.api.client.IHyracksClientConnection;
+import org.apache.hyracks.api.job.JobId;
+import org.apache.hyracks.http.api.IServletRequest;
+import org.apache.hyracks.http.api.IServletResponse;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import io.netty.handler.codec.http.FullHttpRequest;
+import io.netty.handler.codec.http.HttpMethod;
+import io.netty.handler.codec.http.HttpResponseStatus;
+
+public class QueryCancellationServletTest {
+
+    @Test
+    public void testDelete() throws Exception {
+        // Creates a query cancellation servlet.
+        QueryCancellationServlet cancellationServlet = new QueryCancellationServlet(new ConcurrentHashMap<>(),
+                new String[] { "/" });
+        // Adds mocked Hyracks client connection into the servlet context.
+        IHyracksClientConnection mockHcc = mock(IHyracksClientConnection.class);
+        cancellationServlet.ctx().put(ServletConstants.HYRACKS_CONNECTION_ATTR, mockHcc);
+        // Adds a query context into the servlet context.
+        IStatementExecutorContext queryCtx = new StatementExecutorContext();
+        cancellationServlet.ctx().put(ServletConstants.RUNNING_QUERIES_ATTR, queryCtx);
+
+        // Tests the case that query is not in the map.
+        IServletRequest mockRequest = mockRequest("1");
+        IServletResponse mockResponse = mock(IServletResponse.class);
+        cancellationServlet.handle(mockRequest, mockResponse);
+        verify(mockResponse, times(1)).setStatus(HttpResponseStatus.NOT_FOUND);
+
+        // Tests the case that query is in the map.
+        queryCtx.put("1", new JobId(1));
+        cancellationServlet.handle(mockRequest, mockResponse);
+        verify(mockResponse, times(1)).setStatus(HttpResponseStatus.OK);
+
+        // Tests the case the client_context_id is not provided.
+        mockRequest = mockRequest(null);
+        cancellationServlet.handle(mockRequest, mockResponse);
+        verify(mockResponse, times(1)).setStatus(HttpResponseStatus.BAD_REQUEST);
+
+        // Tests the case that the job cancellation hit some exception from Hyracks.
+        queryCtx.put("2", new JobId(2));
+        Mockito.doThrow(new Exception()).when(mockHcc).cancelJob(any());
+        mockRequest = mockRequest("2");
+        cancellationServlet.handle(mockRequest, mockResponse);
+        verify(mockResponse, times(1)).setStatus(HttpResponseStatus.INTERNAL_SERVER_ERROR);
+    }
+
+    private IServletRequest mockRequest(String clientContextId) {
+        IServletRequest mockRequest = mock(IServletRequest.class);
+        FullHttpRequest mockHttpRequest = mock(FullHttpRequest.class);
+        when(mockRequest.getHttpRequest()).thenReturn(mockHttpRequest);
+        when(mockHttpRequest.method()).thenReturn(HttpMethod.DELETE);
+        if (clientContextId != null) {
+            when(mockRequest.getParameter("client_context_id")).thenReturn(clientContextId);
+        }
+        return mockRequest;
+    }
+}

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/343cd457/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/CancellationTestExecutor.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/CancellationTestExecutor.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/CancellationTestExecutor.java
new file mode 100644
index 0000000..97101ba
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/CancellationTestExecutor.java
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.asterix.test.common;
+
+import java.io.InputStream;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.apache.asterix.common.utils.Servlets;
+import org.apache.asterix.testframework.context.TestCaseContext;
+import org.apache.asterix.testframework.xml.TestCase;
+import org.apache.http.HttpResponse;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.client.methods.RequestBuilder;
+import org.junit.Assert;
+
+public class CancellationTestExecutor extends TestExecutor {
+
+    private final ExecutorService executor = Executors.newSingleThreadExecutor();
+
+    @Override
+    public InputStream executeQueryService(String str, TestCaseContext.OutputFormat fmt, URI uri,
+            List<TestCase.CompilationUnit.Parameter> params, boolean jsonEncoded, boolean cancellable)
+            throws Exception {
+        String clientContextId = UUID.randomUUID().toString();
+        if (cancellable) {
+            setParam(params, "client_context_id", clientContextId);
+        }
+        Callable<InputStream> query = () -> {
+            try {
+                return CancellationTestExecutor.super.executeQueryService(str, fmt, uri, params, jsonEncoded, true);
+            } catch (Exception e) {
+                e.printStackTrace();
+                throw e;
+            }
+        };
+        Future<InputStream> future = executor.submit(query);
+        if (cancellable) {
+            Thread.sleep(20);
+            // Cancels the query request while the query is executing.
+            int rc = cancelQuery(getEndpoint(Servlets.RUNNING_REQUESTS), params);
+            Assert.assertTrue(rc == 200 || rc == 404);
+        }
+        InputStream inputStream = future.get();
+        // Since the current cancellation (i.e., abort) implementation is based on thread.interrupt and we did not
+        // track if all task threads are terminated or not, a timed wait here can reduce false alarms.
+        // TODO(yingyi): investigate if we need synchronized cancellation.
+        Thread.sleep(50);
+        return inputStream;
+    }
+
+    // Cancels a submitted query through the cancellation REST API.
+    private int cancelQuery(URI uri, List<TestCase.CompilationUnit.Parameter> params) throws Exception {
+        HttpUriRequest method = constructDeleteMethodUrl(uri, params);
+        HttpResponse response = executeHttpRequest(method);
+        return response.getStatusLine().getStatusCode();
+    }
+
+    // Constructs a HTTP DELETE request.
+    private HttpUriRequest constructDeleteMethodUrl(URI uri, List<TestCase.CompilationUnit.Parameter> otherParams) {
+        RequestBuilder builder = RequestBuilder.delete(uri);
+        for (TestCase.CompilationUnit.Parameter param : otherParams) {
+            builder.addParameter(param.getName(), param.getValue());
+        }
+        builder.setCharset(StandardCharsets.UTF_8);
+        return builder.build();
+    }
+}

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/343cd457/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/ResultExtractor.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/ResultExtractor.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/ResultExtractor.java
index 3531211..53a0f6c 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/ResultExtractor.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/ResultExtractor.java
@@ -19,7 +19,6 @@
 package org.apache.asterix.test.common;
 
 import java.io.InputStream;
-import java.io.StringWriter;
 import java.nio.charset.Charset;
 import java.util.Iterator;
 import java.util.logging.Logger;
@@ -60,6 +59,8 @@ public class ResultExtractor {
             switch (field) {
                 case "requestID":
                     break;
+                case "clientContextID":
+                    break;
                 case "signature":
                     break;
                 case "status":
@@ -106,7 +107,7 @@ public class ResultExtractor {
                     }
                     break;
                 default:
-                    throw new AsterixException(field + "unanticipated field");
+                    throw new AsterixException(field + " unanticipated field");
             }
         }
 

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/343cd457/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/TestExecutor.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/TestExecutor.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/TestExecutor.java
index 7765572..608547c 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/TestExecutor.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/TestExecutor.java
@@ -423,7 +423,6 @@ public class TestExecutor {
             String exceptionMsg;
             try {
                 // First try to parse the response for a JSON error response.
-
                 ObjectMapper om = new ObjectMapper();
                 JsonNode result = om.readTree(errorBody);
                 String[] errors = { result.get("error-code").asText(), result.get("summary").asText(),
@@ -457,6 +456,11 @@ public class TestExecutor {
 
     public InputStream executeQueryService(String str, OutputFormat fmt, URI uri,
             List<CompilationUnit.Parameter> params, boolean jsonEncoded) throws Exception {
+        return executeQueryService(str, fmt, uri, params, jsonEncoded, false);
+    }
+
+    protected InputStream executeQueryService(String str, OutputFormat fmt, URI uri,
+            List<CompilationUnit.Parameter> params, boolean jsonEncoded, boolean cancellable) throws Exception {
         setParam(params, "format", fmt.mimeType());
         HttpUriRequest method = jsonEncoded ? constructPostMethodJson(str, uri, "statement", params)
                 : constructPostMethodUrl(str, uri, "statement", params);
@@ -830,7 +834,7 @@ public class TestExecutor {
                     }
                     final URI uri = getEndpoint(Servlets.QUERY_SERVICE);
                     if (DELIVERY_IMMEDIATE.equals(delivery)) {
-                        resultStream = executeQueryService(statement, fmt, uri, params, true);
+                        resultStream = executeQueryService(statement, fmt, uri, params, true, true);
                         resultStream = ResultExtractor.extract(resultStream);
                     } else {
                         String handleVar = getHandleVariable(statement);

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/343cd457/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/APIExecutionTest.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/APIExecutionTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/APIExecutionTest.java
index 82f90ec..dd87455 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/APIExecutionTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/APIExecutionTest.java
@@ -21,6 +21,7 @@ package org.apache.asterix.test.runtime;
 
 import java.util.Collection;
 
+import org.apache.asterix.test.common.TestExecutor;
 import org.apache.asterix.testframework.context.TestCaseContext;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
@@ -38,7 +39,7 @@ public class APIExecutionTest {
 
     @BeforeClass
     public static void setUp() throws Exception {
-        LangExecutionUtil.setUp(TEST_CONFIG_FILE_NAME);
+        LangExecutionUtil.setUp(TEST_CONFIG_FILE_NAME, new TestExecutor());
     }
 
     @AfterClass

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/343cd457/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/AqlExecutionFullParallelismIT.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/AqlExecutionFullParallelismIT.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/AqlExecutionFullParallelismIT.java
index 8df93ba..6cc5a9c 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/AqlExecutionFullParallelismIT.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/AqlExecutionFullParallelismIT.java
@@ -21,6 +21,7 @@ package org.apache.asterix.test.runtime;
 
 import java.util.Collection;
 
+import org.apache.asterix.test.common.TestExecutor;
 import org.apache.asterix.testframework.context.TestCaseContext;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
@@ -38,7 +39,7 @@ public class AqlExecutionFullParallelismIT {
 
     @BeforeClass
     public static void setUp() throws Exception {
-        LangExecutionUtil.setUp(TEST_CONFIG_FILE_NAME);
+        LangExecutionUtil.setUp(TEST_CONFIG_FILE_NAME, new TestExecutor());
     }
 
     @AfterClass

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/343cd457/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/AqlExecutionLessParallelismIT.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/AqlExecutionLessParallelismIT.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/AqlExecutionLessParallelismIT.java
index 2f3c395..dc03626 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/AqlExecutionLessParallelismIT.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/AqlExecutionLessParallelismIT.java
@@ -21,6 +21,7 @@ package org.apache.asterix.test.runtime;
 
 import java.util.Collection;
 
+import org.apache.asterix.test.common.TestExecutor;
 import org.apache.asterix.testframework.context.TestCaseContext;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
@@ -38,7 +39,7 @@ public class AqlExecutionLessParallelismIT {
 
     @BeforeClass
     public static void setUp() throws Exception {
-        LangExecutionUtil.setUp(TEST_CONFIG_FILE_NAME);
+        LangExecutionUtil.setUp(TEST_CONFIG_FILE_NAME, new TestExecutor());
     }
 
     @AfterClass

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/343cd457/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/AqlExecutionTest.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/AqlExecutionTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/AqlExecutionTest.java
index 4169a07..abc9f2f 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/AqlExecutionTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/AqlExecutionTest.java
@@ -21,6 +21,7 @@ package org.apache.asterix.test.runtime;
 
 import java.util.Collection;
 
+import org.apache.asterix.test.common.TestExecutor;
 import org.apache.asterix.testframework.context.TestCaseContext;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
@@ -38,7 +39,7 @@ public class AqlExecutionTest {
 
     @BeforeClass
     public static void setUp() throws Exception {
-        LangExecutionUtil.setUp(TEST_CONFIG_FILE_NAME);
+        LangExecutionUtil.setUp(TEST_CONFIG_FILE_NAME, new TestExecutor());
     }
 
     @AfterClass

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/343cd457/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/ClusterStateDefaultParameterTest.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/ClusterStateDefaultParameterTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/ClusterStateDefaultParameterTest.java
index 86a9639..17e88a6 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/ClusterStateDefaultParameterTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/ClusterStateDefaultParameterTest.java
@@ -25,6 +25,7 @@ import java.net.HttpURLConnection;
 import java.net.URL;
 import java.util.Collection;
 
+import org.apache.asterix.test.common.TestExecutor;
 import org.apache.asterix.testframework.context.TestCaseContext;
 import org.junit.AfterClass;
 import org.junit.Assert;
@@ -43,7 +44,7 @@ public class ClusterStateDefaultParameterTest {
 
     @BeforeClass
     public static void setUp() throws Exception {
-        LangExecutionUtil.setUp(TEST_CONFIG_FILE_NAME);
+        LangExecutionUtil.setUp(TEST_CONFIG_FILE_NAME, new TestExecutor());
     }
 
     @AfterClass

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/343cd457/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/ClusterStateExecutionFullParallelismTest.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/ClusterStateExecutionFullParallelismTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/ClusterStateExecutionFullParallelismTest.java
index b7b4312..e428c93 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/ClusterStateExecutionFullParallelismTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/ClusterStateExecutionFullParallelismTest.java
@@ -21,6 +21,7 @@ package org.apache.asterix.test.runtime;
 
 import java.util.Collection;
 
+import org.apache.asterix.test.common.TestExecutor;
 import org.apache.asterix.testframework.context.TestCaseContext;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
@@ -38,7 +39,7 @@ public class ClusterStateExecutionFullParallelismTest {
 
     @BeforeClass
     public static void setUp() throws Exception {
-        LangExecutionUtil.setUp(TEST_CONFIG_FILE_NAME);
+        LangExecutionUtil.setUp(TEST_CONFIG_FILE_NAME, new TestExecutor());
     }
 
     @AfterClass

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/343cd457/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/ClusterStateExecutionLessParallelismTest.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/ClusterStateExecutionLessParallelismTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/ClusterStateExecutionLessParallelismTest.java
index 9516d7d..346ae2f 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/ClusterStateExecutionLessParallelismTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/ClusterStateExecutionLessParallelismTest.java
@@ -21,6 +21,7 @@ package org.apache.asterix.test.runtime;
 
 import java.util.Collection;
 
+import org.apache.asterix.test.common.TestExecutor;
 import org.apache.asterix.testframework.context.TestCaseContext;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
@@ -38,7 +39,7 @@ public class ClusterStateExecutionLessParallelismTest {
 
     @BeforeClass
     public static void setUp() throws Exception {
-        LangExecutionUtil.setUp(TEST_CONFIG_FILE_NAME);
+        LangExecutionUtil.setUp(TEST_CONFIG_FILE_NAME, new TestExecutor());
     }
 
     @AfterClass

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/343cd457/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/LangExecutionUtil.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/LangExecutionUtil.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/LangExecutionUtil.java
index 0e6be0f..7c2e472 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/LangExecutionUtil.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/LangExecutionUtil.java
@@ -19,6 +19,8 @@
 
 package org.apache.asterix.test.runtime;
 
+import static org.apache.hyracks.control.common.utils.ThreadDumpHelper.takeDumpJSON;
+
 import java.io.BufferedReader;
 import java.io.File;
 import java.io.IOException;
@@ -35,7 +37,7 @@ import org.apache.asterix.test.common.TestExecutor;
 import org.apache.asterix.testframework.context.TestCaseContext;
 import org.apache.commons.lang.SystemUtils;
 import org.apache.commons.lang3.StringUtils;
-import org.junit.Assert;
+import org.apache.hyracks.control.common.utils.ThreadDumpHelper;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 
@@ -52,12 +54,13 @@ public class LangExecutionUtil {
     private static final boolean cleanupOnStart = true;
     private static final boolean cleanupOnStop = true;
     private static final List<String> badTestCases = new ArrayList<>();
-    private static final TestExecutor testExecutor = new TestExecutor();
+    private static TestExecutor testExecutor;
 
     private static TestLibrarian librarian;
     private static final int repeat = Integer.getInteger("test.repeat", 1);
 
-    public static void setUp(String configFile) throws Exception {
+    public static void setUp(String configFile, TestExecutor executor) throws Exception {
+        testExecutor = executor;
         File outdir = new File(PATH_ACTUAL);
         outdir.mkdirs();
         List<ILibraryManager> libraryManagers = ExecutionTestUtil.setUp(cleanupOnStart, configFile);
@@ -70,16 +73,20 @@ public class LangExecutionUtil {
     }
 
     public static void tearDown() throws Exception {
-        // Check whether there are leaked open run file handles.
-        checkRunFileLeaks();
-
-        TestLibrarian.removeLibraryDir();
-        ExecutionTestUtil.tearDown(cleanupOnStop);
-        ExecutionTestUtil.integrationUtil.removeTestStorageFiles();
-        if (!badTestCases.isEmpty()) {
-            System.out.println("The following test cases left some data");
-            for (String testCase : badTestCases) {
-                System.out.println(testCase);
+        try {
+            // Check whether there are leaked open run file handles.
+            checkOpenRunFileLeaks();
+            // Check whether there are leaked threads.
+            checkThreadLeaks();
+        } finally {
+            TestLibrarian.removeLibraryDir();
+            ExecutionTestUtil.tearDown(cleanupOnStop);
+            ExecutionTestUtil.integrationUtil.removeTestStorageFiles();
+            if (!badTestCases.isEmpty()) {
+                System.out.println("The following test cases left some data");
+                for (String testCase : badTestCases) {
+                    System.out.println(testCase);
+                }
             }
         }
     }
@@ -128,7 +135,18 @@ public class LangExecutionUtil {
         }
     }
 
-    private static void checkRunFileLeaks() throws IOException {
+    private static void checkThreadLeaks() throws IOException {
+        String threadDump = ThreadDumpHelper.takeDumpJSON(ManagementFactory.getThreadMXBean());
+        // Currently we only do sanity check for threads used in the execution engine.
+        // Later we should check if there are leaked storage threads as well.
+        if (threadDump.contains("Operator") || threadDump.contains("SuperActivity") || threadDump
+                .contains("PipelinedPartition")) {
+            System.out.print(threadDump);
+            throw new AssertionError("There are leaked threads in the execution engine.");
+        }
+    }
+
+    private static void checkOpenRunFileLeaks() throws IOException {
         if (SystemUtils.IS_OS_WINDOWS) {
             return;
         }
@@ -142,7 +160,22 @@ public class LangExecutionUtil {
                 .exec(new String[] { "bash", "-c", "lsof -p " + processId + "|grep waf|wc -l" });
         try (BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream()))) {
             int runFileCount = Integer.parseInt(reader.readLine().trim());
-            Assert.assertTrue(runFileCount == 0);
+            if (runFileCount != 0) {
+                System.out.print(takeDumpJSON(ManagementFactory.getThreadMXBean()));
+                outputLeakedOpenFiles(processId);
+                throw new AssertionError("There are " + runFileCount + " leaked run files.");
+            }
+        }
+    }
+
+    private static void outputLeakedOpenFiles(String processId) throws IOException {
+        Process process = Runtime.getRuntime()
+                .exec(new String[] { "bash", "-c", "lsof -p " + processId + "|grep waf" });
+        try (BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream()))) {
+            String line;
+            while ((line = reader.readLine()) != null) {
+                System.err.println(line);
+            }
         }
     }
 }

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/343cd457/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/SqlppExecutionFullParallelismIT.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/SqlppExecutionFullParallelismIT.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/SqlppExecutionFullParallelismIT.java
index f2372ed..3d93c36 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/SqlppExecutionFullParallelismIT.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/SqlppExecutionFullParallelismIT.java
@@ -21,6 +21,7 @@ package org.apache.asterix.test.runtime;
 
 import java.util.Collection;
 
+import org.apache.asterix.test.common.TestExecutor;
 import org.apache.asterix.testframework.context.TestCaseContext;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
@@ -38,7 +39,7 @@ public class SqlppExecutionFullParallelismIT {
 
     @BeforeClass
     public static void setUp() throws Exception {
-        LangExecutionUtil.setUp(TEST_CONFIG_FILE_NAME);
+        LangExecutionUtil.setUp(TEST_CONFIG_FILE_NAME, new TestExecutor());
     }
 
     @AfterClass

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/343cd457/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/SqlppExecutionLessParallelismIT.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/SqlppExecutionLessParallelismIT.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/SqlppExecutionLessParallelismIT.java
index c3cc058..53b068e 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/SqlppExecutionLessParallelismIT.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/SqlppExecutionLessParallelismIT.java
@@ -21,6 +21,7 @@ package org.apache.asterix.test.runtime;
 
 import java.util.Collection;
 
+import org.apache.asterix.test.common.TestExecutor;
 import org.apache.asterix.testframework.context.TestCaseContext;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
@@ -38,7 +39,7 @@ public class SqlppExecutionLessParallelismIT {
 
     @BeforeClass
     public static void setUp() throws Exception {
-        LangExecutionUtil.setUp(TEST_CONFIG_FILE_NAME);
+        LangExecutionUtil.setUp(TEST_CONFIG_FILE_NAME, new TestExecutor());
     }
 
     @AfterClass

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/343cd457/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/SqlppExecutionTest.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/SqlppExecutionTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/SqlppExecutionTest.java
index f19ebbf..8ec1fe7 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/SqlppExecutionTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/SqlppExecutionTest.java
@@ -20,6 +20,7 @@ package org.apache.asterix.test.runtime;
 
 import java.util.Collection;
 
+import org.apache.asterix.test.common.TestExecutor;
 import org.apache.asterix.testframework.context.TestCaseContext;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
@@ -37,7 +38,7 @@ public class SqlppExecutionTest {
 
     @BeforeClass
     public static void setUp() throws Exception {
-        LangExecutionUtil.setUp(TEST_CONFIG_FILE_NAME);
+        LangExecutionUtil.setUp(TEST_CONFIG_FILE_NAME, new TestExecutor());
     }
 
     @AfterClass

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/343cd457/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/SqlppExecutionWithCancellationTest.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/SqlppExecutionWithCancellationTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/SqlppExecutionWithCancellationTest.java
new file mode 100644
index 0000000..edf5741
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/SqlppExecutionWithCancellationTest.java
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.asterix.test.runtime;
+
+import java.util.Collection;
+
+import org.apache.asterix.test.common.CancellationTestExecutor;
+import org.apache.asterix.testframework.context.TestCaseContext;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+
+/**
+ * Runs the SQL++ runtime tests with a cancellation request for each read-only query.
+ */
+@RunWith(Parameterized.class)
+public class SqlppExecutionWithCancellationTest {
+    protected static final String TEST_CONFIG_FILE_NAME = "asterix-build-configuration.xml";
+    private static int numCancelledQueries = 0;
+
+    @BeforeClass
+    public static void setUp() throws Exception {
+        LangExecutionUtil.setUp(TEST_CONFIG_FILE_NAME, new CancellationTestExecutor());
+    }
+
+    @AfterClass
+    public static void tearDown() throws Exception {
+        System.err.println(numCancelledQueries + " queries have been cancelled during the test.");
+        try {
+            // Makes sure that there are queries that have indeed been cancelled during the test.
+            Assert.assertTrue(numCancelledQueries > 0);
+        } finally {
+            LangExecutionUtil.tearDown();
+        }
+    }
+
+    @Parameters(name = "SqlppExecutionWithCancellationTest {index}: {0}")
+    public static Collection<Object[]> tests() throws Exception {
+        return LangExecutionUtil.tests("only_sqlpp.xml", "testsuite_sqlpp.xml");
+    }
+
+    protected TestCaseContext tcCtx;
+
+    public SqlppExecutionWithCancellationTest(TestCaseContext tcCtx) {
+        this.tcCtx = tcCtx;
+    }
+
+    @Test
+    public void test() throws Exception {
+        try {
+            LangExecutionUtil.test(tcCtx);
+        } catch (Exception e) {
+            Throwable cause = getRootCause(e);
+            String errorMsg = cause.getMessage();
+            if (errorMsg.startsWith("HYR0025") // Expected, "HYR0025" means a user cancelled the query.
+                    || errorMsg.contains("\"status\": ") // Expected, "status" results for cancelled queries can change.
+                    || errorMsg.contains("reference count = 1") // not expected, but is a false alarm.
+                    || errorMsg.contains("pinned and file is being closed") // not expected, but maybe a false alarm.
+            // happens after the test query: big_object_load_20M.
+            ) {
+                numCancelledQueries++;
+            } else {
+                // Re-throw other kinds of exceptions.
+                throw e;
+            }
+        }
+    }
+
+    // Finds the root cause of Throwable.
+    private Throwable getRootCause(Throwable e) {
+        Throwable current = e;
+        Throwable cause = e.getCause();
+        while (cause != null) {
+            Throwable nextCause = current.getCause();
+            current = cause;
+            cause = nextCause;
+        }
+        return current;
+    }
+}

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/343cd457/asterixdb/asterix-app/src/test/resources/runtimets/queries/types/any-object/any-object.2.query.aql
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries/types/any-object/any-object.2.query.aql b/asterixdb/asterix-app/src/test/resources/runtimets/queries/types/any-object/any-object.2.query.aql
index e7208ca..7d619b4 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/queries/types/any-object/any-object.2.query.aql
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries/types/any-object/any-object.2.query.aql
@@ -18,4 +18,5 @@
  */
 
 for $x in dataset Metadata.Datatype
+where $x.DataverseName = "test"
 return $x;
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/343cd457/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/types/any-object/any-object.2.query.sqlpp
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/types/any-object/any-object.2.query.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/types/any-object/any-object.2.query.sqlpp
index afaa5b1..cf72d38 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/types/any-object/any-object.2.query.sqlpp
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/types/any-object/any-object.2.query.sqlpp
@@ -17,4 +17,6 @@
  * under the License.
  */
 
-select element x from Metadata.Datatype as x;
+select element x
+from Metadata.Datatype as x
+where x.DataverseName = "test";


Mime
View raw message