package edu.washington.escience.myria.operator;

import java.util.Iterator;
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.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<Integer> childrenWithData;

  /**
   * Iterator into {@link #childrenWithData}. Global to fairly get data from children.
   */
  private transient Iterator<Integer> itr;

  /**
   * @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 {
    for (int i = 0; i < childrenWithData.size(); i++) {
      Integer childIdx = itr.next();
      if (!itr.hasNext()) {
        // simulate circular linked list
        itr = childrenWithData.iterator();
      }

      Operator child = getChild(childIdx);
      if (child.eos()) {
        itr.remove();
        continue;
      }
      TupleBatch tb = child.nextReady();
      if (tb != null) {
        return tb;
      }
    }

    return null;
  }

  @Override
  public void init(final ImmutableMap<String, Object> execEnvVars) throws DbException {
    Objects.requireNonNull(getChildren());
    Preconditions.checkArgument(getNumChildren() > 0);

    childrenWithData = new LinkedList<Integer>();
    int i = 0;
    for (Operator child : getChildren()) {
      Preconditions.checkNotNull(child);
      Preconditions.checkArgument(getSchema().equals(child.getSchema()));
      childrenWithData.add(i++);
    }
    itr = childrenWithData.iterator();
  }
}

@@ -46,35 +46,8 @@
 r;%0A%0A
-import java.util.Iterator;%0A
 impo
@@ -623,176 +623,34 @@
 ist%3C
-Integer%3E childrenWithData;%0A%0A  /**%0A   * Iterator into %7B@link #childrenWithData%7D. Global to fairly get data from children.%0A   */%0A  private transient Iterator%3CInteger%3E itr
+Operator%3E childrenWithData
 ;%0A%0A 
@@ -929,261 +929,468 @@
     
-for (int i = 0; i %3C childrenWithData.size(); i++) %7B%0A      Integ
+/*%0A     * If this variable gets to 0, it means that we've checked ev
 er
+y
  child
-Idx = itr.next();%0A      if (!itr.hasNext()) %7B%0A        // simulate circular linked list%0A        itr = childrenWithData.iterator();%0A      %7D%0A%0A      Operator child = getChild(childIdx
+ that could have data and none currently do.%0A     * At that point, we return null and sleep until one of those children gets data.%0A     */%0A    int uncheckedChildren = childrenWithData.size();%0A%0A    while (!childrenWithData.isEmpty()) %7B%0A      if (uncheckedChildren == 0) %7B%0A        return null;%0A      %7D%0A      uncheckedChildren--;%0A      Operator child = childrenWithData.removeFirst(
 );%0A
+%0A
     
@@ -1414,30 +1414,8 @@
 ) %7B%0A
-        itr.remove();%0A
     
@@ -1432,24 +1432,25 @@
 ue;%0A      %7D%0A
+%0A
       TupleB
@@ -1474,24 +1474,100 @@
 extReady();%0A
+      if (!child.eos()) %7B%0A        childrenWithData.addLast(child);%0A      %7D%0A%0A
       if (tb
@@ -1737,16 +1737,44 @@
 on %7B%0A   
+ final Operator%5B%5D children =
  Objects
@@ -1841,24 +1841,23 @@
 ent(
-getNumChildren()
+children.length
  %3E 0
@@ -1902,33 +1902,19 @@
 ist%3C
-Integer%3E();%0A    int i = 0
+Operator%3E()
 ;%0A  
@@ -2100,11 +2100,13 @@
 add(
-i++
+child
 );%0A 
@@ -2114,47 +2114,8 @@
   %7D%0A
-    itr = childrenWithData.iterator();%0A
   %7D%0A

fix semantics