Merge branch 'slxu-fix-network-backpressure' into slxu-kill-query-at-aborted-downloading

Conflicts:
	src/edu/washington/escience/myria/parallel/Server.java

Author: slxu

.travis.yml

@@ -12,7 +12,7 @@ before_install:
 - sh travis/before_install.sh
 - export TERM=dumb
 after_success:
-- gradle jacocoTestReport coveralls
+- ./gradlew jacocoTestReport coveralls
 
 
 

build.gradle

@@ -4,7 +4,7 @@ apply plugin: "java"
 apply plugin: "eclipse"
 /* Coveralls with JaCoCo */
 apply plugin: 'jacoco'
-apply plugin: 'coveralls'
+apply plugin: 'com.github.kt3k.coveralls'
 
 /* Set up group and version info for myria */
 archivesBaseName = "myria"
@@ -26,7 +26,7 @@ buildscript {
   }
 
   dependencies {
-    classpath 'org.kt3k.gradle.plugin:coveralls-gradle-plugin:0.4.0'
+    classpath 'org.kt3k.gradle.plugin:coveralls-gradle-plugin:2.0.0'
   }
 }
 /* Tell gradle where the source code is located. */

src/edu/washington/escience/myria/coordinator/catalog/WorkerCatalog.java

@@ -22,8 +22,8 @@
 
 /**
  * This class is intended to store the configuration and catalog information for a Myria worker.
- * 
- * 
+ *
+ *
  */
 public final class WorkerCatalog {
   /** The logger for this class. */
@@ -33,7 +33,7 @@ public final class WorkerCatalog {
    * @param filename the path to the SQLite database storing the worker catalog.
    * @return a fresh WorkerCatalog fitting the specified description.
    * @throws CatalogException if there is an error creating the database or the file already exists.
-   * 
+   *
    *           TODO add some sanity checks to the filename?
    */
   public static WorkerCatalog create(final String filename) throws CatalogException {
@@ -46,7 +46,7 @@ public static WorkerCatalog create(final String filename) throws CatalogExceptio
    * @param overwrite specifies whether to overwrite an existing WorkerCatalog.
    * @return a fresh WorkerCatalog fitting the specified description.
    * @throws CatalogException if the database already exists, or there is an error creating it.
-   * 
+   *
    *           TODO add some sanity checks to the filename?
    */
   public static WorkerCatalog create(final String filename, final boolean overwrite) throws CatalogException {
@@ -61,12 +61,12 @@ public static WorkerCatalog create(final String filename, final boolean overwrit
   }
 
   /**
-   * 
+   *
    * @param catalogFile a File object pointing to the SQLite database that will store the WorkerCatalog. If catalogFile
    *          is null, this creates an in-memory SQLite database.
    * @return a fresh WorkerCatalog fitting the specified description.
    * @throws CatalogException if there is an error opening the database.
-   * 
+   *
    *           TODO add some sanity checks to the filename?
    */
   private static WorkerCatalog createFromFile(final File catalogFile) throws CatalogException {
@@ -89,36 +89,36 @@ private static WorkerCatalog createFromFile(final File catalogFile) throws Catal
       sqliteConnection.exec("DROP TABLE IF EXISTS configuration");
       sqliteConnection.exec(
           "CREATE TABLE configuration (\n"
-        + "    key STRING UNIQUE NOT NULL,\n"
-        + "    value STRING NOT NULL);");
+              + "    key STRING UNIQUE NOT NULL,\n"
+              + "    value STRING NOT NULL);");
       sqliteConnection.exec("DROP TABLE IF EXISTS masters");
       sqliteConnection.exec(
           "CREATE TABLE masters (\n"
-        + "    master_id INTEGER PRIMARY KEY ASC,\n"
-        + "    host_port STRING NOT NULL);");
+              + "    master_id INTEGER PRIMARY KEY ASC,\n"
+              + "    host_port STRING NOT NULL);");
       sqliteConnection.exec("DROP TABLE IF EXISTS workers");
       sqliteConnection.exec(
           "CREATE TABLE workers (\n"
-        + "    worker_id INTEGER PRIMARY KEY ASC,\n"
-        + "    host_port STRING NOT NULL);");
+              + "    worker_id INTEGER PRIMARY KEY ASC,\n"
+              + "    host_port STRING NOT NULL);");
       sqliteConnection.exec("DROP TABLE IF EXISTS relations");
       sqliteConnection.exec(
           "CREATE TABLE relations (\n"
-        + "    relation_id INTEGER PRIMARY KEY ASC,\n"
-        + "    relation_name STRING NOT NULL UNIQUE);");
+              + "    relation_id INTEGER PRIMARY KEY ASC,\n"
+              + "    relation_name STRING NOT NULL UNIQUE);");
       sqliteConnection.exec("DROP TABLE IF EXISTS relation_schema");
       sqliteConnection.exec(
           "CREATE TABLE relation_schema (\n"
-        + "    relation_id INTEGER NOT NULL REFERENCES relations(relation_id),\n"
-        + "    col_index INTEGER NOT NULL,\n"
-        + "    col_name STRING,\n"
-        + "    col_type STRING NOT NULL);");
+              + "    relation_id INTEGER NOT NULL REFERENCES relations(relation_id),\n"
+              + "    col_index INTEGER NOT NULL,\n"
+              + "    col_name STRING,\n"
+              + "    col_type STRING NOT NULL);");
       sqliteConnection.exec("DROP TABLE IF EXISTS shards");
       sqliteConnection.exec(
           "CREATE TABLE shards (\n"
-        + "    stored_relation_id INTEGER NOT NULL REFERENCES stored_relations(stored_relation_id),\n"
-        + "    shard_index INTEGER NOT NULL,\n"
-        + "    location STRING NOT NULL);");
+              + "    stored_relation_id INTEGER NOT NULL REFERENCES stored_relations(stored_relation_id),\n"
+              + "    shard_index INTEGER NOT NULL,\n"
+              + "    location STRING NOT NULL);");
       sqliteConnection.exec("COMMIT TRANSACTION");
       /* @formatter:on*/
     } catch (final SQLiteException e) {
@@ -147,7 +147,7 @@ private static WorkerCatalog createFromFile(final File catalogFile) throws Catal
   /**
    * @return a fresh WorkerCatalog fitting the specified description.
    * @throws CatalogException if there is an error opening the database.
-   * 
+   *
    *           TODO add some sanity checks to the filename?
    */
   public static WorkerCatalog createInMemory() throws CatalogException {
@@ -156,12 +156,12 @@ public static WorkerCatalog createInMemory() throws CatalogException {
 
   /**
    * Opens the worker catalog stored as a SQLite database in the specified file.
-   * 
+   *
    * @param filename the path to the SQLite database storing the catalog.
    * @return an initialized WorkerCatalog object ready to be used for experiments.
    * @throws FileNotFoundException if the given file does not exist.
    * @throws CatalogException if there is an error connecting to the database.
-   * 
+   *
    *           TODO add some sanity checks to the filename?
    */
   public static WorkerCatalog open(final String filename) throws FileNotFoundException, CatalogException {
@@ -195,7 +195,7 @@ public static WorkerCatalog open(final String filename) throws FileNotFoundExcep
 
   /**
    * Not publicly accessible.
-   * 
+   *
    * @param sqliteConnection connection to the SQLite database that stores the WorkerCatalog.
    * @throws SQLiteException if there is an error turning on foreign keys.
    */
@@ -208,7 +208,7 @@ private WorkerCatalog(final SQLiteConnection sqliteConnection) throws SQLiteExce
 
   /**
    * Adds a master using the specified host and port to the WorkerCatalog.
-   * 
+   *
    * @param hostPortString specifies the path to the master in the format "host:port"
    * @return this WorkerCatalog
    * @throws CatalogException if the hostPortString is invalid or there is a database exception.
@@ -237,13 +237,13 @@ public WorkerCatalog addMaster(final String hostPortString) throws CatalogExcept
 
   /**
    * Adds the metadata for a relation into the WorkerCatalog.
-   * 
+   *
    * @param name the name of the relation.
    * @param schema the schema of the relation.
    * @throws CatalogException if the relation is already in the WorkerCatalog or there is an error in the database.
-   * 
+   *
    *           TODO if we use SQLite in a multithreaded way this will need to be revisited.
-   * 
+   *
    */
   public void addRelationMetadata(final String name, final Schema schema) throws CatalogException {
     Objects.requireNonNull(name);
@@ -291,7 +291,7 @@ public void addRelationMetadata(final String name, final Schema schema) throws C
 
   /**
    * Adds a worker using the specified host and port to the WorkerCatalog.
-   * 
+   *
    * @param workerId specifies the global identifier of this worker.
    * @param hostPortString specifies the path to the worker in the format "host:port".
    * @return this WorkerCatalog.
@@ -333,7 +333,6 @@ public WorkerCatalog addWorkers(final Map<String, String> id2HostPortString) thr
       throw new CatalogException("WorkerCatalog is closed.");
     }
     try {
-      /* Just used to verify that hostPortString is legal */
       final SQLiteStatement statement =
           sqliteConnection.prepare("INSERT INTO workers(worker_id, host_port) VALUES(?,?);", false);
       sqliteConnection.exec("BEGIN TRANSACTION");
@@ -379,7 +378,7 @@ public void close() {
   /**
    * Extract the value of a particular configuration parameter from the database. Returns null if the parameter is not
    * configured.
-   * 
+   *
    * @param key the name of the configuration parameter.
    * @return the value of the configuration parameter, or null if that configuration is not supported.
    * @throws CatalogException if there is an error in the backing database.
@@ -489,7 +488,7 @@ public Map<Integer, SocketInfo> getWorkers() throws CatalogException {
   /**
    * Extract the value of a particular configuration parameter from the database. Returns null if the parameter is not
    * configured.
-   * 
+   *
    * @param key the name of the configuration parameter.
    * @param value the value of the configuration parameter.
    * @throws CatalogException if there is an error in the backing database.
@@ -515,7 +514,7 @@ public void setConfigurationValue(final String key, final String value) throws C
 
   /**
    * Set all the configuration values in the provided map in a single transaction.
-   * 
+   *
    * @param entries the value of the configuration parameter.
    * @throws CatalogException if there is an error in the backing database.
    */

src/edu/washington/escience/myria/expression/CastExpression.java

@@ -92,7 +92,7 @@ public CastExpression(final ExpressionOperator left, final ExpressionOperator ri
   public Type getOutputType(final ExpressionOperatorParameter parameters) {
     final Type castFrom = getLeft().getOutputType(parameters);
     final Type castTo = getRight().getOutputType(parameters);
-    Preconditions.checkArgument(getCastType(castFrom, castTo) != CastType.UNSUPPORTED,
+    Preconditions.checkArgument((castFrom == castTo) || getCastType(castFrom, castTo) != CastType.UNSUPPORTED,
         "Cast from %s to %s is not supported.", castFrom, castTo);
     return castTo;
   }

src/edu/washington/escience/myria/operator/agg/MultiGroupByAggregate.java

@@ -212,8 +212,10 @@ private TupleBatch getResultBatch() throws DbException {
     TupleBatchBuffer curGroupAggs = new TupleBatchBuffer(aggSchema);
     for (int row = 0; row < curGroupKeys.numTuples(); ++row) {
       Aggregator[] rowAggs = groupAggs.get(row);
-      for (int i = 0; i < rowAggs.length; ++i) {
-        rowAggs[i].getResult(curGroupAggs, i);
+      int curCol = 0;
+      for (Aggregator rowAgg : rowAggs) {
+        rowAgg.getResult(curGroupAggs, curCol);
+        curCol += rowAgg.getResultSchema().numColumns();
       }
     }
     TupleBatch aggResults = curGroupAggs.popAny();
@@ -252,10 +254,8 @@ public Schema generateSchema() {
 
     for (AggregatorFactory f : factories) {
       Schema curAggSchema = f.getResultSchema(inputSchema);
-      Preconditions.checkState(curAggSchema.numColumns() == 1, "aggSchema should only have 1 column, not %s",
-          curAggSchema.numColumns());
-      aggTypes.add(curAggSchema.getColumnType(0));
-      aggNames.add(curAggSchema.getColumnName(0));
+      aggTypes.addAll(curAggSchema.getColumnTypes());
+      aggNames.addAll(curAggSchema.getColumnNames());
     }
     aggSchema = new Schema(aggTypes, aggNames);
     return Schema.merge(groupSchema, aggSchema);