refactoring to eliminate parallel flatmap classes

src/edu/washington/escience/myria/Type.java

@@ -40,6 +40,11 @@ public Class<?> toJavaType() {
       return int.class;
     }
 
+    @Override
+    public Class<?> toJavaArrayType() {
+      return int[].class;
+    }
+
     @Override
     public Class<?> toJavaObjectType() {
       return Integer.class;
@@ -80,6 +85,11 @@ public Class<?> toJavaType() {
       return float.class;
     }
 
+    @Override
+    public Class<?> toJavaArrayType() {
+      return float[].class;
+    }
+
     @Override
     public Class<?> toJavaObjectType() {
       return Float.class;
@@ -120,6 +130,11 @@ public Class<?> toJavaType() {
       return double.class;
     }
 
+    @Override
+    public Class<?> toJavaArrayType() {
+      return double[].class;
+    }
+
     @Override
     public Class<?> toJavaObjectType() {
       return Double.class;
@@ -160,6 +175,11 @@ public Class<?> toJavaType() {
       return boolean.class;
     }
 
+    @Override
+    public Class<?> toJavaArrayType() {
+      return boolean[].class;
+    }
+
     @Override
     public Class<?> toJavaObjectType() {
       return Boolean.class;
@@ -200,6 +220,11 @@ public Class<?> toJavaType() {
       return String.class;
     }
 
+    @Override
+    public Class<?> toJavaArrayType() {
+      return String[].class;
+    }
+
     @Override
     public Class<?> toJavaObjectType() {
       return toJavaType();
@@ -240,6 +265,11 @@ public Class<?> toJavaType() {
       return long.class;
     }
 
+    @Override
+    public Class<?> toJavaArrayType() {
+      return long[].class;
+    }
+
     @Override
     public Class<?> toJavaObjectType() {
       return Long.class;
@@ -280,6 +310,11 @@ public Class<?> toJavaType() {
       return DateTime.class;
     }
 
+    @Override
+    public Class<?> toJavaArrayType() {
+      return DateTime[].class;
+    }
+
     @Override
     public Class<?> toJavaObjectType() {
       return toJavaType();
@@ -503,6 +538,11 @@ private static boolean evalOp(final SimplePredicate.Op op, final int compared) {
    */
   public abstract Class<?> toJavaType();
 
+  /**
+   * @return the java array type
+   */
+  public abstract Class<?> toJavaArrayType();
+
   /**
    * @return the non primitive java type
    */

src/edu/washington/escience/myria/api/encoding/ApplyEncoding.java

@@ -11,7 +11,7 @@ public class ApplyEncoding extends UnaryOperatorEncoding<Apply> {
   @Required public List<Expression> emitExpressions;
 
   @Override
-  public Apply construct(ConstructArgs args) {
+  public Apply construct(final ConstructArgs args) {
     return new Apply(null, emitExpressions);
   }
 }

src/edu/washington/escience/myria/api/encoding/OperatorEncoding.java

@@ -43,7 +43,6 @@
   @Type(name = "EOSController", value = EOSControllerEncoding.class),
   @Type(name = "FileScan", value = FileScanEncoding.class),
   @Type(name = "Filter", value = FilterEncoding.class),
-  @Type(name = "FlatteningApply", value = FlatteningApplyEncoding.class),
   @Type(name = "HyperShuffleProducer", value = HyperShuffleProducerEncoding.class),
   @Type(name = "HyperShuffleConsumer", value = HyperShuffleConsumerEncoding.class),
   @Type(name = "IDBController", value = IDBControllerEncoding.class),

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

@@ -46,11 +46,10 @@ public Type getOutputType(final ExpressionOperatorParameter parameters) {
 
   @Override
   public String getJavaString(final ExpressionOperatorParameter parameters) {
-    // TODO: use IntStream when we switch to Java 8
     return new StringBuilder()
-        .append("int[] counter; for (int i = 0; i < (int) (")
+        .append("IntStream.range(0, (")
         .append(getOperand().getJavaString(parameters))
-        .append("); ++i) {\ncounter[i] = i;\n}\nreturn counter;")
+        .append(")).toArray()")
         .toString();
   }
 
@@ -60,13 +59,13 @@ public String getJavaExpressionWithAppend(final ExpressionOperatorParameter para
         .append(Expression.COUNT)
         .append(".appendInt((int) (")
         .append(getOperand().getJavaString(parameters))
-        .append("));\nfor (int i = 0; i < (int) (")
+        .append("));\n")
+        // It would be nice to replace this loop with IntStream.forEach(), but Janino doesn't support lambdas.
+        .append("for (int i = 0; i < (int) (")
         .append(getOperand().getJavaString(parameters))
         .append("); ++i) {\n")
         .append(Expression.RESULT)
-        .append(".putInt(")
-        .append(Expression.COL)
-        .append(", i);\n}")
+        .append(".appendInt(i);\n}")
         .toString();
   }
 

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

@@ -49,10 +49,6 @@ public class Expression implements Serializable {
    * Variable name of row index.
    */
   public static final String ROW = "row";
-  /**
-   * Variable name of row index.
-   */
-  public static final String COL = "col";
   /**
    * Variable name of state.
    */
@@ -120,21 +116,19 @@ public String getJavaExpressionWithAppend(final ExpressionOperatorParameter para
     String appendExpression = rootExpressionOperator.getJavaExpressionWithAppend(parameters);
     if (appendExpression == null) {
       if (isMultivalued()) {
-        String primitiveTypeName = getOutputType(parameters).toJavaType().getName();
+        String primitiveTypeName = getOutputType(parameters).toJavaArrayType().getSimpleName();
         appendExpression =
             new StringBuilder(primitiveTypeName)
-                .append("[] results = ")
+                .append(" results = ")
                 .append(getJavaExpression(parameters))
                 .append(";\n")
                 .append(COUNT)
                 .append(".appendInt(results.length);\n")
                 .append("for (int i = 0; i < results.length; ++i) {\n")
                 .append(RESULT)
-                .append(".put")
+                .append(".append")
                 .append(getOutputType(parameters).getName())
-                .append("(")
-                .append(COL)
-                .append(", results[i]);\n}")
+                .append("(results[i]);\n}")
                 .toString();
       } else {
         appendExpression =
@@ -143,7 +137,7 @@ public String getJavaExpressionWithAppend(final ExpressionOperatorParameter para
                 .append(getOutputType(parameters).getName())
                 .append("(")
                 .append(getJavaExpression(parameters))
-                .append(")")
+                .append(");")
                 .toString();
       }
     }
@@ -195,8 +189,7 @@ public boolean isConstant() {
   }
 
   /**
-   * An expression is "multivalued" when it has a primitive array return type. This is a requirement for being used in
-   * the FlatteningApply operator.
+   * An expression is "multivalued" when it has a primitive array return type.
    *
    * @return if the root expression has a primitive array return type
    */

src/edu/washington/escience/myria/expression/evaluate/ConstantEvaluator.java

@@ -98,13 +98,17 @@ public Object eval() {
   public void eval(
       final ReadableTable tb,
       final int rowIdx,
+      final WritableColumn count,
       final WritableColumn result,
       final ReadableTable state) {
-    throw new UnsupportedOperationException("Should not be here. Should be using eval() instead");
+    throw new UnsupportedOperationException(
+        "Should not be here. Should be using evaluateColumn() instead");
   }
 
   @Override
-  public Column<?> evaluateColumn(final TupleBatch tb) {
-    return new ConstantValueColumn((Comparable<?>) value, type, tb.numTuples());
+  public EvaluatorResult evaluateColumn(final TupleBatch tb) {
+    return new EvaluatorResult(
+        new ConstantValueColumn((Comparable<?>) value, type, tb.numTuples()),
+        new ConstantValueColumn(1, Type.INT_TYPE, tb.numTuples()));
   }
 }

src/edu/washington/escience/myria/expression/evaluate/ExpressionEvalInterface.java

@@ -4,8 +4,8 @@
 import edu.washington.escience.myria.storage.ReadableTable;
 
 /**
- * Interface for evaluating a single {@link edu.washington.escience.myria.expression.Expression} and appending it to a
- * column.
+ * Interface for evaluating a single {@link edu.washington.escience.myria.expression.Expression} and appending the
+ * results to a column, along with a count of results.
  */
 public interface ExpressionEvalInterface {
   /**
@@ -16,13 +16,15 @@ public interface ExpressionEvalInterface {
    * using the rowId provided in {@link edu.washington.escience.myria.expression.VariableExpression}.
    *
    * @param tb a tuple batch
-   * @param rowId the row in the tb that should be used.
-   * @param result the result column that the value should be appended to
+   * @param row index of the row in the tb that should be used
+   * @param count a column storing the number of results returned from this row
+   * @param result a table storing evaluation results
    * @param state optional state that is passed during evaluation
    */
   void evaluate(
       final ReadableTable tb,
-      final int rowId,
+      final int row,
+      final WritableColumn count,
       final WritableColumn result,
       final ReadableTable state);
 }