package edu.washington.escience.myria.parallel;

import edu.washington.escience.myria.TupleBatch;
import edu.washington.escience.myria.operator.Operator;

/**
 * The producer part of the Collect Exchange operator.
 * 
 * The producer actively pushes the tuples generated by the child operator to the paired CollectConsumer.
 * 
 */
public class CollectProducer extends GenericShuffleProducer {

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

  /**
   * @param child the child who provides data for this producer to distribute.
   * @param operatorID destination operator the data goes
   * @param collectConsumerWorkerID destination worker the data goes.
   * */
  public CollectProducer(final Operator child, final ExchangePairID operatorID, final int collectConsumerWorkerID) {
    super(child, new ExchangePairID[] { operatorID }, new int[][] { { 0 } }, new int[] { collectConsumerWorkerID },
        new FixValuePartitionFunction(0), true);
  }

  @Override
  protected TupleBatch[] getTupleBatchPartitions(final TupleBatch tup) {
    return new TupleBatch[] { tup };
  }
}

@@ -42,16 +42,69 @@
 allel;%0A%0A
+import edu.washington.escience.myria.MyriaConstants;%0A
 import e
@@ -947,14 +947,45 @@
 %7B %7B 
-0 %7D %7D,
+MyriaConstants.MASTER_ID %7D %7D,%0A       
  new
@@ -1019,24 +1019,16 @@
 kerID %7D,
-%0A       
  new Fix

Use myria constant for master worker id