added the query kill test

.travis.yml

@@ -1,16 +1,30 @@
+# sudo: false
+# FIXME: Java compiler dies with OOM if run in container
+sudo: required
+dist: trusty
 language: java
 jdk:
 - oraclejdk8
+python:
+- "2.7"
 addons:
-  postgresql: 9.3
+  postgresql: 9.4
 before_install:
 - sh travis/before_install.sh
 - export TERM=dumb
+script:
+- ./gradlew shadowJar check
+- ./myriadeploy/launch_local_cluster > ./myria.log 2>&1 &
+- sleep 60
+- python ~/myria-web/appengine/myria_web_main.py > ./myria-web.log 2>&1 &
+- sleep 60
+- ./integration_tests.py
+after_failure:
+- cat ./myria.log
+- cat ./myria-web.log
+- find . -name '*.stderr' | xargs cat
 after_success:
 - ./gradlew jacocoTestReport coveralls
-
-sudo: false
-
 cache:
   directories:
   - $HOME/.gradle/caches

build.gradle

@@ -282,6 +282,8 @@ shadowJar {
   /* Some of our deps conflict with old versions in Hadoop distro */
   relocate 'com.google.common', 'myriadeps.com.google.common'
   relocate 'javax.ws.rs', 'myriadeps.javax.ws.rs'
+  relocate 'com.fasterxml.jackson', 'myriadeps.com.fasterxml.jackson'
+  relocate 'org.glassfish.jersey', 'myriadeps.org.glassfish.jersey'
 }
 
 /* Coveralls */

integration_tests.py

@@ -0,0 +1,103 @@
+#!/usr/bin/env python
+
+import unittest
+from collections import Counter
+from tempfile import NamedTemporaryFile
+from myria import MyriaConnection, MyriaRelation, MyriaQuery, MyriaError
+
+
+class MyriaTestBase(unittest.TestCase):
+    def setUp(self):
+        connection = MyriaConnection(hostname='localhost', port=8753, execution_url="http://127.0.0.1:8080")
+        MyriaRelation.DefaultConnection = connection
+        self.connection = connection
+
+    def assertListOfDictsEqual(self, left, right):
+        self.assertEqual(Counter([kv for d in left for kv in d.items()]),
+                         Counter([kv for d in right for kv in d.items()]))
+
+class DoWhileTest(MyriaTestBase):
+    def test(self):
+        program = """
+x = [0 as exp, 1 as val];
+do
+  x = [from x emit exp+1 as exp, 2*val as val];
+while [from x emit max(exp) < 5];
+store(x, powersOfTwo);
+"""
+        query = MyriaQuery.submit(program)
+        expected = [{'val': 32, 'exp': 5}]
+
+        self.assertEqual(query.status, 'SUCCESS')
+        self.assertListOfDictsEqual(query.to_dict(), expected)
+
+
+class IngestEmptyQueryTest(MyriaTestBase):
+    def test(self):
+        # Create empty file
+        with NamedTemporaryFile() as f:
+            #TODO change URL to local file
+            program = """
+emptyrelation = load('https://s3-us-west-2.amazonaws.com/bhaynestemp/emptyrelation', csv(schema(foo:string, bar:int)));
+store(emptyrelation, emptyrelation);
+"""
+        expected = []
+        query = MyriaQuery.submit(program)
+        self.assertEqual(query.status, 'SUCCESS')
+        self.assertListOfDictsEqual(query.to_dict(), expected)
+
+
+class UploadAndReplaceDataTest(MyriaTestBase):
+    def test(self):
+        with NamedTemporaryFile() as f:
+            data = "foo,3242\n" \
+                   "bar,321\n"\
+                   "baz,104"
+            f.write(data)
+            #TODO change URL to local file
+            program = """
+uploaddatatest = load('https://s3-us-west-2.amazonaws.com/bhaynestemp/uploaddatatest', csv(schema(s:string, i:int)));
+store(uploaddatatest, uploaddatatest);
+"""
+            expected = [{u's': u'foo', u'i': 3242},
+                        {u's': u'bar', u'i': 321},
+                        {u's': u'baz', u'i': 104}]
+            query = MyriaQuery.submit(program)
+            self.assertEqual(query.status, 'SUCCESS')
+            self.assertListOfDictsEqual(query.to_dict(), expected)
+
+            # Now replace with another relation
+            program = """
+uploaddatatest = load('https://s3-us-west-2.amazonaws.com/bhaynestemp/uploaddatatest2',
+                       csv(schema(s:string, i:int), delimiter='\\t'));
+store(uploaddatatest, uploaddatatest);
+"""
+            expected = [{u's': u'mexico', u'i': 42},
+                        {u's': u'poland', u'i': 12342},
+                        {u's': u'belize', u'i': 802304}]
+            query = MyriaQuery.submit(program)
+            self.assertEqual(query.status, 'SUCCESS')
+            self.assertListOfDictsEqual(query.to_dict(), expected)
+
+class KillQueryTest(MyriaTestBase):
+    def test(self):
+        # This is a long-running program
+        program = """
+x = [0 as exp, 1 as val];
+do
+  x = [from x emit exp+1 as exp, 2*val as val];
+while [from x emit max(exp) < 1000];
+store(x, powersOfTwo);
+"""
+        plan = self.connection.compile_program(program)
+        # We submit with submit_plan so as to not block
+        query = MyriaQuery.submit_plan(plan)
+
+        # We kill the query and make sure it was killed
+        query.kill()
+        self.assertEqual(query.status, 'KILLED')
+
+
+
+if __name__ == '__main__':
+    unittest.main()

src/edu/washington/escience/myria/accessmethod/SQLiteAccessMethod.java

@@ -79,6 +79,7 @@ public void connect(final ConnectionInfo connectionInfo, final Boolean readOnly)
           if (parent != null && !parent.exists()) {
             dbFile.getParentFile().mkdirs();
           }
+          System.err.println("About to create new database file");
           dbFile.createNewFile();
         } catch (IOException e) {
           throw new DbException("Could not create database file" + dbFile.getAbsolutePath(), e);
@@ -97,15 +98,45 @@ public void connect(final ConnectionInfo connectionInfo, final Boolean readOnly)
         sqliteConnection = new SQLiteConnection(new File(sqliteInfo.getDatabaseFilename()));
         sqliteConnection.openReadonly();
         sqliteConnection.setBusyTimeout(SQLiteAccessMethod.DEFAULT_BUSY_TIMEOUT);
+        setJournalMode(sqliteConnection);
       } else {
         sqliteQueue = new SQLiteQueue(new File(sqliteInfo.getDatabaseFilename())).start();
+        setJournalModeAsync(sqliteQueue);
       }
     } catch (final SQLiteException e) {
       LOGGER.error(e.getMessage(), e);
       throw new DbException(e);
     }
   }
 
+  private static void setJournalMode(SQLiteConnection sqliteConnection) throws DbException {
+    Objects.requireNonNull(sqliteConnection);
+    try {
+      sqliteConnection.exec("PRAGMA journal_mode=WAL;");
+    } catch (final SQLiteException e) {
+      LOGGER.error(e.getMessage());
+      throw new DbException(e);
+    }
+  }
+
+  private static void setJournalModeAsync(SQLiteQueue sqliteQueue) throws DbException {
+    Objects.requireNonNull(sqliteQueue);
+    try {
+      sqliteQueue
+          .execute(
+              new SQLiteJob<Object>() {
+                @Override
+                protected Object job(final SQLiteConnection sqliteConnection) throws DbException {
+                  setJournalMode(sqliteConnection);
+                  return null;
+                }
+              })
+          .get();
+    } catch (InterruptedException | ExecutionException e) {
+      throw new DbException(e);
+    }
+  }
+
   @Override
   public void setReadOnly(final Boolean readOnly) throws DbException {
     Objects.requireNonNull(sqliteConnection);

src/edu/washington/escience/myria/daemon/MyriaDriverLauncher.java

@@ -65,7 +65,14 @@ public final class MyriaDriverLauncher {
    */
   public static void main(final String[] args) throws Exception {
     LauncherStatus status = run(args);
-    LOGGER.info("Exit with status " + status);
+    LOGGER.info("Driver launcher exiting with status " + status);
+    if (status == LauncherStatus.FAILED) {
+      System.exit(1);
+    } else if (status == LauncherStatus.FORCE_CLOSED) {
+      System.exit(2);
+    } else {
+      System.exit(0);
+    }
   }
 
   private final REEF reef;
@@ -347,7 +354,6 @@ private LauncherStatus run(final Configuration driverConf) {
     try {
       LOGGER.info("Submitting Myria driver to REEF...");
       reef.submit(driverConf);
-
       synchronized (this) {
         while (!status.isDone()) {
           try {
@@ -363,7 +369,6 @@ private LauncherStatus run(final Configuration driverConf) {
           }
         }
       }
-
       return status;
     } finally {
       reef.close();

src/edu/washington/escience/myria/operator/DbInsert.java

@@ -206,19 +206,6 @@ protected void init(final ImmutableMap<String, Object> execEnvVars) throws DbExc
       throw new DbException("Unable to instantiate DbInsert: connection information unknown");
     }
 
-    if (connectionInfo instanceof SQLiteInfo) {
-      /* Set WAL in the beginning. */
-      final File dbFile = new File(((SQLiteInfo) connectionInfo).getDatabaseFilename());
-      SQLiteConnection conn = new SQLiteConnection(dbFile);
-      try {
-        conn.open(true);
-        conn.exec("PRAGMA journal_mode=WAL;");
-      } catch (SQLiteException e) {
-        e.printStackTrace();
-      }
-      conn.dispose();
-    }
-
     /* open the database connection */
     accessMethod = AccessMethod.of(connectionInfo.getDbms(), connectionInfo, false);
 

src/edu/washington/escience/myria/operator/DbInsertTemp.java

@@ -27,6 +27,12 @@
 import edu.washington.escience.myria.storage.TupleBatch;
 import edu.washington.escience.myria.storage.TupleUtils;
 
+import java.nio.file.Files;
+import java.nio.file.attribute.PosixFileAttributes;
+import java.nio.file.attribute.PosixFileAttributeView;
+import java.nio.file.attribute.PosixFilePermissions;
+import java.io.IOException;
+
 /** A temporary relation that is inserted into the database. */
 public class DbInsertTemp extends AbstractDbInsert {
 
@@ -192,19 +198,6 @@ protected void setupConnection(final ImmutableMap<String, Object> execEnvVars)
       throw new DbException("Unknown connection information.");
     }
 
-    if (connectionInfo instanceof SQLiteInfo) {
-      /* Set WAL in the beginning. */
-      final File dbFile = new File(((SQLiteInfo) connectionInfo).getDatabaseFilename());
-      SQLiteConnection conn = new SQLiteConnection(dbFile);
-      try {
-        conn.open(true);
-        conn.exec("PRAGMA journal_mode=WAL;");
-      } catch (SQLiteException e) {
-        e.printStackTrace();
-      }
-      conn.dispose();
-    }
-
     // Open the database connection.
     accessMethod = AccessMethod.of(connectionInfo.getDbms(), connectionInfo, false);
   }

src/edu/washington/escience/myria/operator/IDBController.java

@@ -366,18 +366,6 @@ public final void init(final ImmutableMap<String, Object> execEnvVars) throws Db
       if (connectionInfo == null) {
         throw new DbException("Unable to instantiate DbInsert: connection information unknown");
       }
-      if (connectionInfo instanceof SQLiteInfo) {
-        /* Set WAL in the beginning. */
-        final File dbFile = new File(((SQLiteInfo) connectionInfo).getDatabaseFilename());
-        SQLiteConnection conn = new SQLiteConnection(dbFile);
-        try {
-          conn.open(true);
-          conn.exec("PRAGMA journal_mode=WAL;");
-        } catch (SQLiteException e) {
-          e.printStackTrace();
-        }
-        conn.dispose();
-      }
       /* open the database connection */
       accessMethod = AccessMethod.of(connectionInfo.getDbms(), connectionInfo, false);
       accessMethod.dropTableIfExists(relationKey);

src/edu/washington/escience/myria/operator/Operator.java

@@ -114,7 +114,7 @@ public ProfilingLogger getProfilingLogger() {
    */
   public PythonFunctionRegistrar getPythonFunctionRegistrar() throws DbException {
     PythonFunctionRegistrar pyFuncRegistrar = null;
-    if (!execEnvVars.containsKey(MyriaConstants.EXEC_ENV_VAR_TEST_MODE)) {
+    if (execEnvVars == null || !execEnvVars.containsKey(MyriaConstants.EXEC_ENV_VAR_TEST_MODE)) {
       if (getLocalSubQuery() instanceof WorkerSubQuery) {
         pyFuncRegistrar =
             ((WorkerSubQuery) getLocalSubQuery()).getWorker().getPythonFunctionRegistrar();

systemtest/edu/washington/escience/myria/accessmethod/JdbcAccessMethodTest.java → test/edu/washington/escience/myria/accessmethod/JdbcAccessMethodTest.java

@@ -1,5 +1,5 @@
 /**
- * 
+ *
  */
 package edu.washington.escience.myria.accessmethod;
 
@@ -18,17 +18,18 @@
 import edu.washington.escience.myria.RelationKey;
 import edu.washington.escience.myria.Schema;
 import edu.washington.escience.myria.Type;
+import edu.washington.escience.myria.operator.BatchTupleSource;
 import edu.washington.escience.myria.operator.DbInsert;
 import edu.washington.escience.myria.operator.DbQueryScan;
 import edu.washington.escience.myria.operator.LeafOperator;
 import edu.washington.escience.myria.operator.TupleRangeSource;
-import edu.washington.escience.myria.operator.BatchTupleSource;
 import edu.washington.escience.myria.storage.TupleBatch;
 import edu.washington.escience.myria.storage.TupleBatchBuffer;
+import edu.washington.escience.myria.util.TestEnvVars;
 import edu.washington.escience.myria.util.TestUtils;
 
 /**
- * 
+ *
  */
 public class JdbcAccessMethodTest {
   /** The logger for this class. */
@@ -41,8 +42,14 @@ public class JdbcAccessMethodTest {
   private JdbcInfo getJdbcInfo(final String dbms) {
     Verify.verify(TestUtils.inTravis(), "This test should only run in Travis");
     /* Return localhost using Travis' default credentials. */
-    return JdbcInfo.of(POSTGRES_DRIVER_CLASS, MyriaConstants.STORAGE_SYSTEM_POSTGRESQL, "localhost", POSTGRES_PORT,
-        POSTGRES_DATABASE_NAME, "postgres", "");
+    return JdbcInfo.of(
+        POSTGRES_DRIVER_CLASS,
+        MyriaConstants.STORAGE_SYSTEM_POSTGRESQL,
+        "localhost",
+        POSTGRES_PORT,
+        POSTGRES_DATABASE_NAME,
+        "postgres",
+        "");
   }
 
   private void testInsertTuplesAndCountThem(final String dbms) throws DbException {
@@ -54,7 +61,8 @@ private void testInsertTuplesAndCountThem(final String dbms) throws DbException
   private void testInsertTuplesAndCountThemWithNull(final String dbms) throws DbException {
     final int expectedNumResults = 250;
     TupleBatchBuffer data =
-        new TupleBatchBuffer(Schema.of(ImmutableList.of(Type.STRING_TYPE), ImmutableList.of("value")));
+        new TupleBatchBuffer(
+            Schema.of(ImmutableList.of(Type.STRING_TYPE), ImmutableList.of("value")));
     for (int i = 0; i < expectedNumResults; i++) {
       if (i % 2 == 0) {
         data.putString(0, "");
@@ -66,7 +74,8 @@ private void testInsertTuplesAndCountThemWithNull(final String dbms) throws DbEx
     doInsert(dbms, source, expectedNumResults);
   }
 
-  private void doInsert(final String dbms, final LeafOperator source, final int expectedNumResults) throws DbException {
+  private void doInsert(final String dbms, final LeafOperator source, final int expectedNumResults)
+      throws DbException {
     Objects.requireNonNull(dbms, "dbms");
 
     /* Connection information */