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 };
  }
}

package edu.washington.escience.myria.parallel;

import edu.washington.escience.myria.MyriaConstants;
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[][] { { MyriaConstants.MASTER_ID } },
        new int[] { collectConsumerWorkerID }, new FixValuePartitionFunction(0), true);
  }

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

Use myria constant for master worker id
Use myria constant for master worker id