package edu.washington.escience.myria.operator;

import java.util.LinkedList;
import java.util.Objects;

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

import edu.washington.escience.myria.DbException;
import edu.washington.escience.myria.Schema;
import edu.washington.escience.myria.storage.TupleBatch;

/**
 * Unions the output of a set of operators without eliminating duplicates.
 * */
public final class UnionAll extends NAryOperator {

  /** Required for Java serialization. */
  private static final long serialVersionUID = 1L;

  /**
   * List of children that have not yet returned EOS.
   */
  private transient LinkedList<Operator> childrenWithData;

  /**
   * @param children the children to be united.
   * */
  public UnionAll(final Operator[] children) {
    super(children);
  }

  @Override
  protected void cleanup() throws DbException {
  }

  @Override
  protected TupleBatch fetchNextReady() throws DbException {
    /*
     * If this variable gets to 0, it means that we've checked every child that could have data and none currently do.
     * At that point, we return null and sleep until one of those children gets data.
     */
    int uncheckedChildren = childrenWithData.size();

    while (!childrenWithData.isEmpty()) {
      if (uncheckedChildren == 0) {
        return null;
      }
      uncheckedChildren--;
      Operator child = childrenWithData.removeFirst();

      if (child.eos()) {
        continue;
      }

      TupleBatch tb = child.nextReady();
      if (!child.eos()) {
        childrenWithData.addLast(child);
      }

      if (tb != null) {
        return tb;
      }
    }

    return null;
  }

  @Override
  public void init(final ImmutableMap<String, Object> execEnvVars) throws DbException {
    final Operator[] children = Objects.requireNonNull(getChildren());
    Preconditions.checkArgument(children.length > 0);

    childrenWithData = new LinkedList<Operator>();
    for (Operator child : getChildren()) {
      Preconditions.checkNotNull(child);

      if (!getSchema().compatible(child.getSchema())) {
        throw new DbException("Incompatible input schema");
      }
      childrenWithData.add(child);
    }
  }

  @Override
  public Schema generateSchema() {
    Operator[] children = getChildren();
    if (children == null) {
      return null;
    }
    if (children[0] == null) {
      return null;
    }
    return children[0].getSchema();
  }
}

@@ -71,16 +71,39 @@
 edList;%0A
+import java.util.List;%0A
 import j
@@ -446,18 +446,16 @@
 icates.%0A
- *
  */%0Apubl
@@ -596,16 +596,138 @@
  = 1L;%0A%0A
+  /** The column names from the first child, whose names we pass on. */%0A  private List%3CString%3E outputColumnNames = null;%0A%0A
   /**%0A  
@@ -900,18 +900,16 @@
 ited.%0A  
- *
  */%0A  pu
@@ -1784,18 +1784,167 @@
     
-return tb;
+if (tb.getSchema().getColumnNames().equals(outputColumnNames)) %7B%0A          return tb;%0A        %7D else %7B%0A          return tb.rename(outputColumnNames);%0A        %7D
 %0A   
@@ -2145,16 +2145,28 @@
 ildren()
+, %22children%22
 );%0A    P
@@ -2211,16 +2211,122 @@
 ngth %3E 0
+, %22UnionAll requires at least one child%22);%0A    outputColumnNames = children%5B0%5D.getSchema().getColumnNames(
 );%0A%0A    
@@ -2453,16 +2453,25 @@
 ll(child
+, %22child%22
 );%0A%0A    
@@ -2472,21 +2472,44 @@
 %0A%0A      
-if (!
+Preconditions.checkArgument(
 getSchem
@@ -2545,79 +2545,95 @@
 a())
-) %7B%0A        throw new DbException(%22Incompatible input schema%22);%0A      %7D
+, %22Child schema %25s is incompatible with %25s%22,%0A          child.getSchema(), getSchema());
 %0A   

repair output column names if necessary