package edu.washington.escience.myria.operator;

import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.List;

import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;

import edu.washington.escience.myria.DbException;
import edu.washington.escience.myria.Schema;
import edu.washington.escience.myria.TupleBatch;
import edu.washington.escience.myria.Type;
import edu.washington.escience.myria.column.Column;
import edu.washington.escience.myria.expression.Expression;
import edu.washington.escience.myria.expression.evaluate.GenericEvaluator;
import edu.washington.escience.myria.expression.evaluate.TupleEvaluator;

/**
 * Generic apply operator.
 */
public class Apply extends UnaryOperator {
  /***/
  private static final long serialVersionUID = 1L;

  /**
   * List of expressions that will be used to create the output.
   */
  private ImmutableList<Expression> emitExpressions;

  /**
   * One evaluator for each expression in {@link #emitExpressions}.
   */
  private ArrayList<GenericEvaluator> emitEvaluators;

  /**
   * @return the {@link #emitExpressions}
   */
  protected ImmutableList<Expression> getEmitExpressions() {
    return emitExpressions;
  }

  /**
   * @return the {@link #emitEvaluators}
   */
  public ArrayList<GenericEvaluator> getEmitEvaluators() {
    return emitEvaluators;
  }

  /**
   * 
   * @param child child operator that data is fetched from
   * @param emitExpressions expression that created the output
   */
  public Apply(final Operator child, final List<Expression> emitExpressions) {
    super(child);
    if (emitExpressions != null) {
      setEmitExpressions(emitExpressions);
    }
  }

  /**
   * @param emitExpressions the emit expressions for each column
   */
  private void setEmitExpressions(final List<Expression> emitExpressions) {
    this.emitExpressions = ImmutableList.copyOf(emitExpressions);
  }

  @Override
  protected TupleBatch fetchNextReady() throws DbException, InvocationTargetException {
    Operator child = getChild();

    if (child.eoi() || getChild().eos()) {
      return null;
    }

    TupleBatch tb = child.nextReady();
    if (tb == null) {
      return null;
    }

    List<Column<?>> output = Lists.newLinkedList();
    for (GenericEvaluator evaluator : emitEvaluators) {
      output.add(evaluator.evaluateColumn(tb));
    }
    return new TupleBatch(getSchema(), output);
  }

  @Override
  protected void init(final ImmutableMap<String, Object> execEnvVars) throws DbException {
    Preconditions.checkNotNull(emitExpressions);

    Schema inputSchema = getChild().getSchema();

    emitEvaluators = new ArrayList<>();
    emitEvaluators.ensureCapacity(emitExpressions.size());
    for (Expression expr : emitExpressions) {
      GenericEvaluator evaluator = new GenericEvaluator(expr, inputSchema, null);
      if (evaluator.needsCompiling()) {
        evaluator.compile();
      }
      emitEvaluators.add(evaluator);
    }
  }

  /**
   * @param evaluators the evaluators to set
   */
  public void setEvaluators(final ArrayList<GenericEvaluator> evaluators) {
    emitEvaluators = evaluators;
  }

  @Override
  public Schema generateSchema() {
    if (emitExpressions == null) {
      return null;
    }
    Operator child = getChild();
    if (child == null) {
      return null;
    }
    Schema childSchema = child.getSchema();
    if (childSchema == null) {
      return null;
    }

    ImmutableList.Builder<Type> typesBuilder = ImmutableList.builder();
    ImmutableList.Builder<String> namesBuilder = ImmutableList.builder();

    for (TupleEvaluator evaluator : emitEvaluators) {
      evaluator.setInputSchema(childSchema);
      typesBuilder.add(evaluator.getOutputType());
      namesBuilder.add(evaluator.getOutputName());
    }
    return new Schema(typesBuilder.build(), namesBuilder.build());
  }
}

@@ -3029,24 +3029,84 @@
 ();%0A      %7D%0A
+      Preconditions.checkArgument(!evaluator.needsState());%0A
       emitEv

Make sure state is not required for expressions in normal apply