public class BufferJoinKeyExtractor implements MapFunction<Tuple, Tuple3<Tuple, Integer, Tuple>> {

	private int[] keyPos;
	private Tuple3<Tuple, Integer, Tuple> outT;
	private Tuple defaultKey = new Tuple(1);

	public BufferJoinKeyExtractor(int inputId, int[] keyPos) {
		this.keyPos = keyPos;
		this.outT = new Tuple3<>(null, inputId, null);
	}

	@Override
	public Tuple3<Tuple, Integer, Tuple> map(Tuple value) throws Exception {

		if(keyPos.length > 0) {
			outT.f0 = value.get(keyPos);
		}
		else {
			outT.f0 = defaultKey;
		}
		outT.f2 = value;
		return outT;
	}
} @SuppressWarnings("unused")
public class CoGroupBufferReducer extends RichGroupReduceFunction<Tuple3<Tuple, Integer, Tuple>, Tuple> {

	private static final Logger LOG = LoggerFactory.getLogger(CoGroupBufferReducer.class);

	private FlowNode flowNode;
	private CoGroupBufferReduceStreamGraph streamGraph;
	private CoGroupBufferInGate groupSource;
	private FlinkFlowProcess currentProcess;

	private boolean calledPrepare;
	private long processBeginTime;

	public CoGroupBufferReducer() {}

	public CoGroupBufferReducer(FlowNode flowNode) {
		this.flowNode = flowNode;
	}

	@Override
	public void open(Configuration config) {

		this.calledPrepare = false;

		try {

			currentProcess = new FlinkFlowProcess(FlinkConfigConverter.toHadoopConfig(config), getRuntimeContext(), flowNode.getID());

			Set<FlowElement> sources = flowNode.getSourceElements();
			if(sources.size() != 1) {
				throw new RuntimeException("FlowNode for CoGroupBufferReducer may only have a single CoGroup source");
			}
			FlowElement sourceElement = sources.iterator().next();
			if(!(sourceElement instanceof CoGroup)) {
				throw new RuntimeException("Source of CoGroupBufferReducer must be a CoGroup");
			}
			CoGroup source = (CoGroup)sourceElement;

			streamGraph = new CoGroupBufferReduceStreamGraph( currentProcess, flowNode, source );
			groupSource = this.streamGraph.getGroupSource();

			for( Duct head : streamGraph.getHeads() ) {
				LOG.info("sourcing from: " + ((ElementDuct) head).getFlowElement());
			}

			for( Duct tail : streamGraph.getTails() ) {
				LOG.info("sinking to: " + ((ElementDuct) tail).getFlowElement());
			}
		}
		catch( Throwable throwable ) {

			if (throwable instanceof CascadingException) {
				throw (CascadingException) throwable;
			}

			throw new FlowException("internal error during CoGroupBufferReducer configuration", throwable);
		}

	}

	@Override
	public void reduce(Iterable<Tuple3<Tuple, Integer, Tuple>> input, Collector<Tuple> output) throws Exception {

		this.streamGraph.setTupleCollector(output);

		if(! this.calledPrepare) {
			this.streamGraph.prepare();
			this.calledPrepare = true;

			this.groupSource.start(this.groupSource);

			processBeginTime = System.currentTimeMillis();
			currentProcess.increment( SliceCounters.Process_Begin_Time, processBeginTime );
		}

		try {
			this.groupSource.run(input.iterator());
		}
		catch( OutOfMemoryError error ) {
			throw error;
		}
		catch( Throwable throwable ) {
			if( throwable instanceof CascadingException ) {
				throw (CascadingException) throwable;
			}

			throw new FlowException( "internal error during CoGroupBufferReducer execution", throwable );
		}

	}

	@Override
	public void close() {

		try {
			if( this.calledPrepare) {
				this.groupSource.complete(this.groupSource);
				this.streamGraph.cleanup();
			}
		}
		finally {
			if( currentProcess != null ) {

				long processEndTime = System.currentTimeMillis();
				currentProcess.increment( SliceCounters.Process_End_Time, processEndTime );
				currentProcess.increment( SliceCounters.Process_Duration, processEndTime - processBeginTime );
			}

			String message = "flow node id: " + flowNode.getID();
			logMemory( LOG, message + ", mem on close" );
			logCounters( LOG, message + ", counter:", currentProcess );
		}
	}

} @SuppressWarnings("unused")
public class CoGroupReducer extends RichGroupReduceFunction<Tuple2<Tuple, Tuple[]>, Tuple> {

	private static final Logger LOG = LoggerFactory.getLogger(CoGroupReducer.class);

	private FlowNode flowNode;
	private CoGroupReduceStreamGraph streamGraph;
	private CoGroupInGate groupSource;
	private FlinkFlowProcess currentProcess;

	private boolean calledPrepare;
	private long processBeginTime;

	public CoGroupReducer() {}

	public CoGroupReducer(FlowNode flowNode) {
		this.flowNode = flowNode;
	}

	@Override
	public void open(Configuration config) {

		this.calledPrepare = false;

		try {

			currentProcess = new FlinkFlowProcess(FlinkConfigConverter.toHadoopConfig(config), getRuntimeContext(), flowNode.getID());

			Set<FlowElement> sources = flowNode.getSourceElements();
			if(sources.size() != 1) {
				throw new RuntimeException("FlowNode for CoGroupReducer may only have a single source");
			}
			FlowElement sourceElement = sources.iterator().next();
			if(!(sourceElement instanceof CoGroup)) {
				throw new RuntimeException("Source of CoGroupReducer must be a CoGroup");
			}
			CoGroup source = (CoGroup)sourceElement;

			streamGraph = new CoGroupReduceStreamGraph( currentProcess, flowNode, source );
			groupSource = this.streamGraph.getGroupSource();

			for( Duct head : streamGraph.getHeads() ) {
				LOG.info("sourcing from: " + ((ElementDuct) head).getFlowElement());
			}

			for( Duct tail : streamGraph.getTails() ) {
				LOG.info("sinking to: " + ((ElementDuct) tail).getFlowElement());
			}
		}
		catch( Throwable throwable ) {

			if (throwable instanceof CascadingException) {
				throw (CascadingException) throwable;
			}

			throw new FlowException("internal error during CoGroupReducer configuration", throwable);
		}

	}

	@Override
	public void reduce(Iterable<Tuple2<Tuple, Tuple[]>> input, Collector<Tuple> output) throws Exception {

		this.streamGraph.setTupleCollector(output);

		if(! this.calledPrepare) {
			this.streamGraph.prepare();
			this.calledPrepare = true;

			this.groupSource.start(this.groupSource);

			this.processBeginTime = System.currentTimeMillis();
			currentProcess.increment( SliceCounters.Process_Begin_Time, processBeginTime );
		}

		try {
			this.groupSource.run(input.iterator());
		}
		catch( OutOfMemoryError error ) {
			throw error;
		}
		catch( Throwable throwable ) {
			if( throwable instanceof CascadingException ) {
				throw (CascadingException) throwable;
			}

			throw new FlowException( "internal error during CoGroupReducer execution", throwable );
		}

	}

	@Override
	public void close() {

		try {
			if( this.calledPrepare) {
				this.groupSource.complete(this.groupSource);

				this.streamGraph.cleanup();
			}
		}
		finally {
			if( currentProcess != null ) {
				long processEndTime = System.currentTimeMillis();
				currentProcess.increment( SliceCounters.Process_End_Time, processEndTime );
				currentProcess.increment( SliceCounters.Process_Duration, processEndTime - processBeginTime );
			}

			String message = "flow node id: " + flowNode.getID();
			logMemory( LOG, message + ", mem on close" );
			logCounters( LOG, message + ", counter:", currentProcess );
		}
	}

} public class TupleAppendOuterJoiner extends RichJoinFunction<Tuple2<Tuple, Tuple[]>, Tuple, Tuple2<Tuple, Tuple[]>> {

	private int tupleListPos;
	private int tupleListSize;
	private Fields inputFields;
	private Fields keyFields;

	private transient Tuple2<Tuple, Tuple[]> outT;

	public TupleAppendOuterJoiner(int tupleListPos, int tupleListSize, Fields inputFields, Fields keyFields) {
		this.tupleListPos = tupleListPos;
		this.tupleListSize = tupleListSize;
		this.inputFields = inputFields;
		this.keyFields = keyFields;
	}

	@Override
	public void open(Configuration config) {
		this.outT = new Tuple2<>(null, new Tuple[tupleListSize]);
	}

	@Override
	public Tuple2<Tuple, Tuple[]> join(Tuple2<Tuple, Tuple[]> leftT, Tuple rightT) throws Exception {
		if(leftT == null) {
			outT.f0 = rightT.get(inputFields, keyFields);
			outT.f1[tupleListPos] = rightT;

			return outT;
		}
		else {
			leftT.f1[tupleListPos] = rightT;

			return leftT;
		}
	}

} public class TupleOuterJoiner extends RichJoinFunction<Tuple, Tuple, Tuple2<Tuple, Tuple[]>> {

	private int tupleListSize;
	private Fields inputFieldsLeft;
	private Fields keyFieldsLeft;
	private Fields inputFieldsRight;
	private Fields keyFieldsRight;

	private transient Tuple2<Tuple, Tuple[]> outT;

	public TupleOuterJoiner(int tupleListSize,
							Fields inputFieldsLeft, Fields keyFieldsLeft,
							Fields inputFieldsRight, Fields keyFieldsRight) {
		this.tupleListSize = tupleListSize;
		this.inputFieldsLeft = inputFieldsLeft;
		this.keyFieldsLeft = keyFieldsLeft;
		this.inputFieldsRight = inputFieldsRight;
		this.keyFieldsRight = keyFieldsRight;
	}

	@Override
	public void open(Configuration config) {
		this.outT = new Tuple2<>(null, new Tuple[tupleListSize]);
	}

	@Override
	public Tuple2<Tuple, Tuple[]> join(Tuple leftT, Tuple rightT) throws Exception {

		if(leftT == null) {
			outT.f0 = rightT.get(inputFieldsRight, keyFieldsRight);
		}
		else {
			outT.f0 = leftT.get(inputFieldsLeft, keyFieldsLeft);
		}

		outT.f1[0] = leftT;
		outT.f1[1] = rightT;

		return outT;
	}

} @SuppressWarnings("unused")
public class GroupByReducer extends RichGroupReduceFunction<Tuple, Tuple> {

	private static final Logger LOG = LoggerFactory.getLogger(GroupByReducer.class);

	private FlowNode flowNode;
	private GroupByStreamGraph streamGraph;
	private GroupByInGate groupSource;
	private FlinkFlowProcess currentProcess;

	private boolean calledPrepare;
	private long processBeginTime;

	public GroupByReducer() {}

	public GroupByReducer(FlowNode flowNode) {
		this.flowNode = flowNode;
	}

	@Override
	public void open(Configuration config) {

		this.calledPrepare = false;

		try {

			currentProcess = new FlinkFlowProcess(FlinkConfigConverter.toHadoopConfig(config), getRuntimeContext(), flowNode.getID());

			Set<FlowElement> sources = flowNode.getSourceElements();
			if(sources.size() != 1) {
				throw new RuntimeException("FlowNode for GroupByReducer may only have a single source");
			}
			FlowElement sourceElement = sources.iterator().next();
			if(!(sourceElement instanceof GroupBy)) {
				throw new RuntimeException("Source of GroupByReducer must be a GroupBy");
			}
			GroupBy source = (GroupBy)sourceElement;

			streamGraph = new GroupByStreamGraph( currentProcess, flowNode, source );
			groupSource = this.streamGraph.getGroupSource();

			for( Duct head : streamGraph.getHeads() ) {
				LOG.info("sourcing from: " + ((ElementDuct) head).getFlowElement());
			}

			for( Duct tail : streamGraph.getTails() ) {
				LOG.info("sinking to: " + ((ElementDuct) tail).getFlowElement());
			}
		}
		catch( Throwable throwable ) {

			if( throwable instanceof CascadingException) {
				throw (CascadingException) throwable;
			}

			throw new FlowException( "internal error during GroupByReducer configuration", throwable );
		}

	}

	@Override
	public void reduce(Iterable<Tuple> input, Collector<Tuple> output) throws Exception {

		this.streamGraph.setTupleCollector(output);

		if(! this.calledPrepare) {
			this.streamGraph.prepare();
			this.calledPrepare = true;

			this.groupSource.start(this.groupSource);

			processBeginTime = System.currentTimeMillis();
			currentProcess.increment( SliceCounters.Process_Begin_Time, processBeginTime );
		}


		try {
			this.groupSource.run(input.iterator());
		}
		catch( OutOfMemoryError error ) {
			throw error;
		}
		catch( Throwable throwable ) {

			if( throwable instanceof CascadingException ) {
				throw (CascadingException) throwable;
			}

			throw new FlowException( "internal error during GroupByReducer execution", throwable );
		}
	}


	@Override
	public void close() {

		try {
			if( this.calledPrepare) {
				this.groupSource.complete(this.groupSource);
				this.streamGraph.cleanup();
			}
		}
		finally {
			if( currentProcess != null ) {
				long processEndTime = System.currentTimeMillis();
				currentProcess.increment( SliceCounters.Process_End_Time, processEndTime );
				currentProcess.increment( SliceCounters.Process_Duration, processEndTime - processBeginTime );
			}

			String message = "flow node id: " + flowNode.getID();
			logMemory( LOG, message + ", mem on close" );
			logCounters( LOG, message + ", counter:", currentProcess );
		}
	}

} public class NaryHashJoinJoiner extends RichFlatJoinFunction<Tuple2<Tuple, Tuple[]>, Tuple, Tuple> {

	private static final Logger LOG = LoggerFactory.getLogger(NaryHashJoinJoiner.class);

	private FlowNode flowNode;
	private int numJoinInputs;

	private transient HashJoinStreamGraph streamGraph;
	private transient FlinkFlowProcess currentProcess;
	private JoinBoundaryInStage sourceStage;
	private transient Tuple[] joinedTuples;

	private transient long processBeginTime;
	private transient boolean prepareCalled;

	public NaryHashJoinJoiner() {}

	public NaryHashJoinJoiner(FlowNode flowNode, int numJoinInputs) {
		this.flowNode = flowNode;
		this.numJoinInputs = numJoinInputs;
	}

	@Override
	public void open(Configuration config) {

		try {

			joinedTuples = new Tuple[numJoinInputs];

			currentProcess = new FlinkFlowProcess(FlinkConfigConverter.toHadoopConfig(config), getRuntimeContext(), flowNode.getID());

			Set<FlowElement> sources = flowNode.getSourceElements();
			// pick one (arbitrary) source
			FlowElement sourceElement = sources.iterator().next();
			if(!(sourceElement instanceof Boundary)) {
				throw new RuntimeException("Source of NaryHashJoinJoiner must be a boundary");
			}

			Boundary source = (Boundary)sourceElement;

			streamGraph = new HashJoinStreamGraph( currentProcess, flowNode, source );
			sourceStage = this.streamGraph.getSourceStage();

			for( Duct head : streamGraph.getHeads() ) {
				LOG.info("sourcing from: " + ((ElementDuct) head).getFlowElement());
			}

			for( Duct tail : streamGraph.getTails() ) {
				LOG.info("sinking to: " + ((ElementDuct) tail).getFlowElement());
			}
		}
		catch( Throwable throwable ) {

			if( throwable instanceof CascadingException) {
				throw (CascadingException) throwable;
			}

			throw new FlowException( "internal error during NaryHashJoinJoiner configuration", throwable );
		}

		this.prepareCalled = false;

	}

	@Override
	public void join(Tuple2<Tuple, Tuple[]> left, Tuple right, Collector<Tuple> output) throws Exception {

		if(!this.prepareCalled) {

			streamGraph.prepare();
			sourceStage.start(null);

			processBeginTime = System.currentTimeMillis();
			currentProcess.increment(SliceCounters.Process_Begin_Time, processBeginTime);
			prepareCalled = true;
		}

		this.streamGraph.setTupleCollector(output);

		System.arraycopy(left.f1, 0, joinedTuples, 0, numJoinInputs - 1);
		joinedTuples[numJoinInputs-1] = right;
		left.f1 = joinedTuples;

		try {
			sourceStage.run(left);
		}
		catch(IOException exception ) {
			throw exception;
		}
		catch( Throwable throwable ) {

			if( throwable instanceof CascadingException ) {
				throw (CascadingException) throwable;
			}

			throw new FlowException( "internal error during NaryHashJoinJoiner execution", throwable );
		}
	}

	@Override
	public void close() {

		try {
			if( this.prepareCalled) {
				this.sourceStage.complete(this.sourceStage);
				this.streamGraph.cleanup();
			}
		}
		finally {
			if( currentProcess != null ) {
				long processEndTime = System.currentTimeMillis();
				currentProcess.increment( SliceCounters.Process_End_Time, processEndTime );
				currentProcess.increment( SliceCounters.Process_Duration, processEndTime - processBeginTime );
			}

			String message = "flow node id: " + flowNode.getID();
			logMemory( LOG, message + ", mem on close" );
			logCounters( LOG, message + ", counter:", currentProcess );
		}
	}

} public class FlinkFlowProcess extends FlowProcess<Configuration> {

	private transient RuntimeContext runtimeContext;
	private Configuration conf;
	private String taskId;

	public FlinkFlowProcess() {
		this.conf = new Configuration();
	}

	public FlinkFlowProcess(Configuration conf) {
		this.conf = conf;
	}

	public FlinkFlowProcess(FlowSession flowSession, Configuration conf) {
		super(flowSession);
		this.conf = conf;
	}

	public FlinkFlowProcess(Configuration conf, RuntimeContext runtimeContext, String taskId) {

		this(conf);
		this.runtimeContext = runtimeContext;
		this.taskId = taskId;
	}

	@Override
	public int getNumProcessSlices() {
		return this.runtimeContext.getNumberOfParallelSubtasks();
	}

	@Override
	public int getCurrentSliceNum() {
		return this.runtimeContext.getIndexOfThisSubtask();
	}

	@Override
	public FlowProcess copyWith(Configuration config) {
		return new FlinkFlowProcess(config, this.runtimeContext, this.taskId);
	}

	@Override
	public Object getProperty( String key ) {
		return this.conf.get(key);
	}

	@Override
	public Collection<String> getPropertyKeys() {
		Set<String> keys = new HashSet<>();

		for( Map.Entry<String, String> entry : this.conf ) {
			keys.add(entry.getKey());
		}

		return Collections.unmodifiableSet( keys );
	}

	@Override
	public Object newInstance(String className) {
		if(className == null || className.isEmpty()) {
			return null;
		}

		try {
			Class clazz = Class.forName(className);
			return InstantiationUtil.instantiate(clazz);
		}
		catch( ClassNotFoundException exception ) {
			throw new CascadingException( "unable to load class: " + className, exception );
		}
	}

	@Override
	public void keepAlive() {
		// Hadoop reports progress
		// Tez doesn't do anything here either...
	}

	@Override
	public void increment(Enum e, long l) {
		increment(EnumStringConverter.enumToGroup(e), EnumStringConverter.enumToCounter(e), l);
	}

	@Override
	public void increment(String group, String counter, long l) {
		if(this.runtimeContext != null) {
			LongCounter flinkCounter = getOrInitCounter(EnumStringConverter.mergeGroupCounter(group, counter));
			flinkCounter.add(l);
		}
	}

	@Override
	public long getCounterValue(Enum e) {
		return getCounterValue(EnumStringConverter.enumToGroup(e), EnumStringConverter.enumToCounter(e));
	}

	@Override
	public long getCounterValue(String group, String counter) {
		if(this.runtimeContext != null) {
			return getOrInitCounter(EnumStringConverter.mergeGroupCounter(group, counter)).getLocalValue();
		}
		else {
			return 0L;
		}
	}

	@Override
	public void setStatus(String s) {
		// Tez doesn't do anything here either
	}

	@Override
	public boolean isCounterStatusInitialized() {
		return this.runtimeContext != null;
	}

	@Override
	public TupleEntryIterator openTapForRead(Tap tap) throws IOException {
		return tap.openForRead( this );
	}

	@Override
	public TupleEntryCollector openTapForWrite(Tap tap) throws IOException {
		return tap.openForWrite( this, null ); // do not honor sinkmode as this may be opened across tasks
	}

	@Override
	public TupleEntryCollector openTrapForWrite(Tap trap) throws IOException {

		if (trap instanceof Hfs) {

			JobConf jobConf = new JobConf(this.getConfigCopy());

			int stepNum = jobConf.getInt( "cascading.flow.step.num", 0 );
			int nodeNum = jobConf.getInt( "cascading.flow.node.num", 0 );

			String partname = String.format( "-%05d-%05d-%05d", stepNum, nodeNum, this.getCurrentSliceNum() );
			jobConf.set( "cascading.tapcollector.partname", "%s%spart" + partname );

			String value = String.format( "attempt_%012d_0000_m_%06d_0", (int) Math.rint( System.currentTimeMillis() ), this.getCurrentSliceNum() );
			jobConf.set( "mapred.task.id", value );
			jobConf.set( "mapreduce.task.id", value );

			return trap.openForWrite( new FlinkFlowProcess( jobConf ), null);
		}
		else {
			throw new UnsupportedOperationException("Only Hfs taps are supported as traps");
		}
	}

	@Override
	public TupleEntryCollector openSystemIntermediateForWrite() throws IOException {
		return null; // Not required for Flink
	}

	@Override
	public Configuration getConfig() {
		return conf;
	}

	@Override
	public Configuration getConfigCopy() {
		return HadoopUtil.copyJobConf(this.conf);
	}

	@Override
	public <C> C copyConfig(C conf) {
		return HadoopUtil.copyJobConf(conf);
	}

	@Override
	public <C> Map<String, String> diffConfigIntoMap(C defaultConfig, C updatedConfig) {

		Map<String, String> newConf = new HashMap<>();
		for(Map.Entry<String,String> e : ((Configuration)updatedConfig)) {
			String key = e.getKey();
			String val = ((Configuration) updatedConfig).get(key);
			String defaultVal = ((Configuration)defaultConfig).get(key);

			// add keys that are different from default
			if(val != null && defaultVal != null && !val.equals(defaultVal)) {
				newConf.put(key, val);
			}
			else if((val == null && defaultVal != null) || (val != null && defaultVal == null)) {
				newConf.put(key, val);
			}
		}

		return newConf;
	}

	@Override
	public Configuration mergeMapIntoConfig(Configuration defaultConfig, Map<String, String> map) {

		Configuration mergedConf = HadoopUtil.copyJobConf(defaultConfig);
		for(String key : map.keySet()) {
			mergedConf.set(key, map.get(key));
		}
		return mergedConf;
	}

	private LongCounter getOrInitCounter(String counterName) {

		LongCounter lc = this.runtimeContext.getLongCounter(counterName);
		if (lc == null) {
			lc = new LongCounter();
			this.runtimeContext.addAccumulator(counterName, lc);
		}
		return lc;
	}

} public class BinaryHashJoinJoiner extends RichFlatJoinFunction<Tuple, Tuple, Tuple> {

	private static final Logger LOG = LoggerFactory.getLogger(BinaryHashJoinJoiner.class);

	private FlowNode flowNode;
	private Fields leftSchema;
	private Fields leftKeyFields;

	private transient HashJoinStreamGraph streamGraph;
	private transient FlinkFlowProcess currentProcess;
	private JoinBoundaryInStage sourceStage;
	private transient Tuple[] joinedTuples;
	private transient Tuple2<Tuple, Tuple[]> joinInput;

	private transient long processBeginTime;
	private transient boolean prepareCalled;

	public BinaryHashJoinJoiner() {}

	public BinaryHashJoinJoiner(FlowNode flowNode, Fields leftSchema, Fields leftKeyFields) {
		this.flowNode = flowNode;
		this.leftSchema = leftSchema;
		this.leftKeyFields = leftKeyFields;
	}

	@Override
	public void open(Configuration config) {

		try {

			this.joinedTuples = new Tuple[2];
			this.joinInput = new Tuple2<>(new Tuple(), this.joinedTuples);

			currentProcess = new FlinkFlowProcess(FlinkConfigConverter.toHadoopConfig(config), getRuntimeContext(), flowNode.getID());

			Set<FlowElement> sources = flowNode.getSourceElements();
			// pick one (arbitrary) source
			FlowElement sourceElement = sources.iterator().next();
			if(!(sourceElement instanceof Boundary)) {
				throw new RuntimeException("Source of BinaryHashJoinJoiner must be a boundary");
			}

			Boundary source = (Boundary)sourceElement;

			streamGraph = new HashJoinStreamGraph( currentProcess, flowNode, source );
			sourceStage = this.streamGraph.getSourceStage();

			for( Duct head : streamGraph.getHeads() ) {
				LOG.info("sourcing from: " + ((ElementDuct) head).getFlowElement());
			}

			for( Duct tail : streamGraph.getTails() ) {
				LOG.info("sinking to: " + ((ElementDuct) tail).getFlowElement());
			}
		}
		catch( Throwable throwable ) {

			if( throwable instanceof CascadingException) {
				throw (CascadingException) throwable;
			}

			throw new FlowException( "internal error during BinaryHashJoinJoiner configuration", throwable );
		}

		this.prepareCalled = false;

	}

	@Override
	public void join(Tuple left, Tuple right, Collector<Tuple> output) throws Exception {

		if(!this.prepareCalled) {

			streamGraph.prepare();
			sourceStage.start(null);

			processBeginTime = System.currentTimeMillis();
			currentProcess.increment(SliceCounters.Process_Begin_Time, processBeginTime);
			prepareCalled = true;
		}

		this.streamGraph.setTupleCollector(output);

		joinInput.f0 = left.get(leftSchema, leftKeyFields);
		joinedTuples[0] = left;
		joinedTuples[1] = right;

		try {
			sourceStage.run(joinInput);
		}
		catch(IOException exception ) {
			throw exception;
		}
		catch( Throwable throwable ) {

			if( throwable instanceof CascadingException ) {
				throw (CascadingException) throwable;
			}

			throw new FlowException( "internal error during BinaryHashJoinJoiner execution", throwable );
		}
	}

	@Override
	public void close() {

		try {
			if( this.prepareCalled) {
				this.sourceStage.complete(this.sourceStage);
				this.streamGraph.cleanup();
			}
		}
		finally {
			if( currentProcess != null ) {
				long processEndTime = System.currentTimeMillis();
				currentProcess.increment( SliceCounters.Process_End_Time, processEndTime );
				currentProcess.increment( SliceCounters.Process_Duration, processEndTime - processBeginTime );
			}

			String message = "flow node id: " + flowNode.getID();
			logMemory( LOG, message + ", mem on close" );
			logCounters( LOG, message + ", counter:", currentProcess );
		}
	}

} public class JoinPrepareMapper implements MapFunction<Tuple, Tuple2<Tuple, Tuple[]>> {

	final private Tuple2<Tuple, Tuple[]> out;
	private final int initialPos;
	final private Fields schema;
	final private Fields keyFields;
	final private Tuple empty = new Tuple();

	public JoinPrepareMapper(int numJoinInputs, Fields schema, Fields keyFields) {
		this(numJoinInputs, 0, schema, keyFields);
	}

	public JoinPrepareMapper(int numJoinInputs, int initialPos, Fields schema, Fields keyFields) {
		this.schema = schema;
		this.keyFields = keyFields;
		this.initialPos = initialPos;

		Tuple[] tupleList = new Tuple[numJoinInputs];
		for(int i=0; i<numJoinInputs-1; i++) {
			tupleList[i] = new Tuple();
		}
		out = new Tuple2<>(null, tupleList);
	}

	@Override
	public Tuple2<Tuple, Tuple[]> map(Tuple tuple) throws Exception {
		if(keyFields == null) {
			out.f0 = empty;
		}
		else {
			out.f0 = tuple.get(schema, keyFields);
		}
		out.f1[initialPos] = tuple;
		return out;
	}

} public class TupleAppendCrosser implements CrossFunction<Tuple2<Tuple, Tuple[]>, Tuple, Tuple2<Tuple, Tuple[]>> {

	private int tupleListPos;

	public TupleAppendCrosser(int tupleListPos) {
		this.tupleListPos = tupleListPos;
	}

	@Override
	public Tuple2<Tuple, Tuple[]> cross(Tuple2<Tuple, Tuple[]> left, Tuple right) throws Exception {

		left.f1[tupleListPos] = right;
		return left;
	}
} public class TupleAppendJoiner implements JoinFunction<Tuple2<Tuple, Tuple[]>, Tuple, Tuple2<Tuple, Tuple[]>> {

	private int tupleListPos;

	public TupleAppendJoiner(int tupleListPos) {
		this.tupleListPos = tupleListPos;
	}

	@Override
	public Tuple2<Tuple, Tuple[]> join(Tuple2<Tuple, Tuple[]> left, Tuple right) throws Exception {

		left.f1[tupleListPos] = right;
		return left;
	}
} @SuppressWarnings("unused")
public class HashJoinMapper extends RichMapPartitionFunction<Tuple2<Tuple, Tuple[]>, Tuple> {

	private static final Logger LOG = LoggerFactory.getLogger(HashJoinMapper.class);

	private FlowNode flowNode;
	private HashJoinMapperStreamGraph streamGraph;
	private JoinBoundaryMapperInStage sourceStage;
	private FlinkFlowProcess currentProcess;

	public HashJoinMapper() {}

	public HashJoinMapper(FlowNode flowNode) {
		this.flowNode = flowNode;
	}

	@Override
	public void open(Configuration config) {

		try {

			currentProcess = new FlinkFlowProcess(FlinkConfigConverter.toHadoopConfig(config), getRuntimeContext(), flowNode.getID());

			Set<FlowElement> sources = flowNode.getSourceElements();
			// pick one (arbitrary) source
			FlowElement sourceElement = sources.iterator().next();
			if(!(sourceElement instanceof Boundary)) {
				throw new RuntimeException("Source of HashJoinMapper must be a boundary");
			}

			Boundary source = (Boundary)sourceElement;

			streamGraph = new HashJoinMapperStreamGraph( currentProcess, flowNode, source );
			sourceStage = this.streamGraph.getSourceStage();

			for( Duct head : streamGraph.getHeads() ) {
				LOG.info("sourcing from: " + ((ElementDuct) head).getFlowElement());
			}

			for( Duct tail : streamGraph.getTails() ) {
				LOG.info("sinking to: " + ((ElementDuct) tail).getFlowElement());
			}
		}
		catch( Throwable throwable ) {

			if( throwable instanceof CascadingException) {
				throw (CascadingException) throwable;
			}

			throw new FlowException( "internal error during HashJoinMapper configuration", throwable );
		}

	}

	@Override
	public void mapPartition(Iterable<Tuple2<Tuple, Tuple[]>> input, Collector<Tuple> output) throws Exception {

		this.streamGraph.setTupleCollector(output);
		streamGraph.prepare();

		long processBeginTime = System.currentTimeMillis();
		currentProcess.increment( SliceCounters.Process_Begin_Time, processBeginTime );

		try {
			try {

				sourceStage.run( input.iterator() );
			}
			catch( OutOfMemoryError error ) {
				throw error;
			}
			catch( IOException exception ) {
				throw exception;
			}
			catch( Throwable throwable ) {

				if( throwable instanceof CascadingException ) {
					throw (CascadingException) throwable;
				}

				throw new FlowException( "internal error during HashJoinMapper execution", throwable );
			}
		}
		finally {
			try {
				streamGraph.cleanup();
			}
			finally {
				long processEndTime = System.currentTimeMillis();
				currentProcess.increment( SliceCounters.Process_End_Time, processEndTime );
				currentProcess.increment( SliceCounters.Process_Duration, processEndTime - processBeginTime );
			}
		}
	}

} @SuppressWarnings("unused")
public class EachMapper extends RichMapPartitionFunction<Tuple, Tuple> {

	private static final Logger LOG = LoggerFactory.getLogger(EachMapper.class);

	private FlowNode flowNode;
	private EachStreamGraph streamGraph;
	private BoundaryInStage sourceStage;
	private FlinkFlowProcess currentProcess;

	public EachMapper() {}

	public EachMapper(FlowNode flowNode) {
		this.flowNode = flowNode;
	}

	@Override
	public void open(Configuration config) {

		try {

			currentProcess = new FlinkFlowProcess(FlinkConfigConverter.toHadoopConfig(config), getRuntimeContext(), flowNode.getID());

			Set<FlowElement> sources = flowNode.getSourceElements();
			if(sources.size() != 1) {
				throw new RuntimeException("FlowNode for EachMapper may only have a single source");
			}

			FlowElement sourceElement = sources.iterator().next();
			if(!(sourceElement instanceof Boundary)) {
				throw new RuntimeException("Source of EachMapper must be a single Boundary");
			}

			Boundary source = (Boundary)sourceElement;

			streamGraph = new EachStreamGraph( currentProcess, flowNode, source );
			sourceStage = this.streamGraph.getSourceStage();

			for( Duct head : streamGraph.getHeads() ) {
				LOG.info("sourcing from: " + ((ElementDuct) head).getFlowElement());
			}

			for( Duct tail : streamGraph.getTails() ) {
				LOG.info("sinking to: " + ((ElementDuct) tail).getFlowElement());
			}

		}
		catch( Throwable throwable ) {

			if( throwable instanceof CascadingException) {
				throw (CascadingException) throwable;
			}

			throw new FlowException( "internal error during EachMapper configuration", throwable );
		}
	}

	@Override
	public void mapPartition(Iterable<Tuple> input, Collector<Tuple> output) throws Exception {

		this.streamGraph.setTupleCollector(output);
		streamGraph.prepare();

		long processBeginTime = System.currentTimeMillis();

		currentProcess.increment( SliceCounters.Process_Begin_Time, processBeginTime );

		try {
			try {
				sourceStage.run( input.iterator() );
			}
			catch( OutOfMemoryError error ) {
				throw error;
			}
			catch( IOException exception ) {
				throw exception;
			}
			catch( Throwable throwable ) {

				if( throwable instanceof CascadingException ) {
					throw (CascadingException) throwable;
				}

				throw new FlowException( "internal error during EachMapper execution", throwable );
			}
		}
		finally {
			try {
				streamGraph.cleanup();
			}
			finally {
				long processEndTime = System.currentTimeMillis();

				currentProcess.increment( SliceCounters.Process_End_Time, processEndTime );
				currentProcess.increment( SliceCounters.Process_Duration, processEndTime - processBeginTime );
			}
		}
	}

} public class TapOutputFormat extends RichOutputFormat<Tuple> implements FinalizeOnMaster {

	private static final long serialVersionUID = 1L;

	private static final Logger LOG = LoggerFactory.getLogger(TapOutputFormat.class);

	private FlowNode flowNode;

	private transient org.apache.hadoop.conf.Configuration config;
	private transient FlinkFlowProcess flowProcess;
	private transient SinkStreamGraph streamGraph;
	private transient SinkBoundaryInStage sourceStage;

	private transient long processBeginTime;

	public TapOutputFormat(FlowNode node) {
		super();
		this.flowNode = node;
	}

	@Override
	public void configure(Configuration config) {

		this.config = FlinkConfigConverter.toHadoopConfig(config);
	}

	@Override
	public void open(int taskNumber, int numTasks) throws IOException {

		this.processBeginTime = System.currentTimeMillis();

		BigInteger numId = new BigInteger(flowNode.getID(), 16);
		String hadoopTaskId = String.format( "attempt_%012d_0000_%s_%06d_0", numId.longValue(), "m", taskNumber );

		this.config.setInt("mapred.task.partition", taskNumber);
		this.config.set("mapred.task.id", hadoopTaskId);

		try {

			flowProcess = new FlinkFlowProcess(this.config, this.getRuntimeContext(), flowNode.getID());

			Set<FlowElement> sources = flowNode.getSourceElements();
			if(sources.size() != 1) {
				throw new RuntimeException("FlowNode for TapOutputFormat may only have a single source");
			}
			FlowElement sourceElement = sources.iterator().next();
			if(!(sourceElement instanceof Boundary)) {
				throw new RuntimeException("Source of TapOutputFormat must be a Boundary");
			}
			Boundary source = (Boundary)sourceElement;

			streamGraph = new SinkStreamGraph( flowProcess, flowNode, source );
			sourceStage = this.streamGraph.getSourceStage();

			for( Duct head : streamGraph.getHeads() ) {
				LOG.info("sourcing from: " + ((ElementDuct) head).getFlowElement());
			}

			for( Duct tail : streamGraph.getTails() ) {
				LOG.info("sinking to: " + ((ElementDuct) tail).getFlowElement());
			}
		}
		catch( Throwable throwable ) {

			if( throwable instanceof CascadingException) {
				throw (CascadingException) throwable;
			}

			throw new FlowException( "internal error during TapOutputFormat configuration", throwable );
		}

		streamGraph.prepare();

	}

	@Override
	public void writeRecord(Tuple t) throws IOException {

		try {
			sourceStage.run( t );
		}
		catch( OutOfMemoryError error ) {
			throw error;
		}
		catch( IOException exception ) {
			throw exception;
		}
		catch( Throwable throwable ) {

			if( throwable instanceof CascadingException ) {
				throw (CascadingException) throwable;
			}

			throw new FlowException( "internal error during TapOutputFormat execution", throwable );
		}
	}

	@Override
	public void close() throws IOException {
		try {
			streamGraph.cleanup();
		}
		finally {
			long processEndTime = System.currentTimeMillis();

			flowProcess.increment( SliceCounters.Process_End_Time, processEndTime );
			flowProcess.increment( SliceCounters.Process_Duration, processEndTime - this.processBeginTime );
		}
	}

	@Override
	public void finalizeGlobal(int parallelism) throws IOException {

		org.apache.hadoop.conf.Configuration config = HadoopUtil.copyConfiguration(this.config);
		Tap tap = this.flowNode.getSinkTaps().iterator().next();

		config.setBoolean(HadoopUtil.CASCADING_FLOW_EXECUTING, false);
		HadoopUtil.setOutputPath(config, new Path(tap.getIdentifier()));

		Hadoop18TapUtil.cleanupJob( config );
	}
} public class TapInputFormat extends RichInputFormat<Tuple, HadoopInputSplit> {

	private static final long serialVersionUID = 1L;

	private static final Logger LOG = LoggerFactory.getLogger(TapInputFormat.class);

	private FlowNode flowNode;

	private transient SourceStreamGraph streamGraph;
	private transient TapSourceStage sourceStage;
	private transient SingleOutBoundaryStage sinkStage;

	private transient FlowProcess flowProcess;
	private transient long processBeginTime;

	private transient org.apache.hadoop.mapred.InputFormat<? extends WritableComparable, ? extends Writable> mapredInputFormat;
	private transient JobConf jobConf;

	public TapInputFormat(FlowNode flowNode) {

		super();
		this.flowNode = flowNode;

	}

	@SuppressWarnings("unchecked")
	@Override
	public void configure(Configuration config) {

		this.jobConf = HadoopUtil.asJobConfInstance(FlinkConfigConverter.toHadoopConfig(config));

		// set the correct class loader
		// not necessary for Flink versions >= 0.10 but we set this anyway to be on the safe side
		jobConf.setClassLoader(this.getClass().getClassLoader());

		this.mapredInputFormat = jobConf.getInputFormat();

		if (this.mapredInputFormat instanceof JobConfigurable) {
			((JobConfigurable) this.mapredInputFormat).configure(jobConf);
		}
	}



	@Override
	public void open(HadoopInputSplit split) throws IOException {

		this.jobConf = split.getJobConf();
		this.flowProcess = new FlinkFlowProcess(this.jobConf, this.getRuntimeContext(), flowNode.getID());

		processBeginTime = System.currentTimeMillis();
		flowProcess.increment( SliceCounters.Process_Begin_Time, processBeginTime );

		try {

			Set<FlowElement> sources = flowNode.getSourceElements();
			if(sources.size() != 1) {
				throw new RuntimeException("FlowNode for TapInputFormat may only have a single source");
			}
			FlowElement sourceElement = sources.iterator().next();
			if(!(sourceElement instanceof Tap)) {
				throw new RuntimeException("Source of TapInputFormat must be a Tap");
			}
			Tap source = (Tap)sourceElement;

			streamGraph = new SourceStreamGraph( flowProcess, flowNode, source );

			sourceStage = this.streamGraph.getSourceStage();
			sinkStage = this.streamGraph.getSinkStage();

			for( Duct head : streamGraph.getHeads() ) {
				LOG.info("sourcing from: " + ((ElementDuct) head).getFlowElement());
			}

			for( Duct tail : streamGraph.getTails() ) {
				LOG.info("sinking to: " + ((ElementDuct) tail).getFlowElement());
			}

		}
		catch( Throwable throwable ) {

			if( throwable instanceof CascadingException) {
				throw (CascadingException) throwable;
			}

			throw new FlowException( "internal error during TapInputFormat configuration", throwable );
		}

		RecordReader<?, ?> recordReader = this.mapredInputFormat.getRecordReader(split.getHadoopInputSplit(), jobConf, new HadoopDummyReporter());

		if (recordReader instanceof Configurable) {
			((Configurable) recordReader).setConf(jobConf);
		}
		else if (recordReader instanceof JobConfigurable) {
			((JobConfigurable) recordReader).configure(jobConf);
		}

		try {
			this.sourceStage.setRecordReader(recordReader);
		} catch(Throwable t) {
			if(t instanceof IOException) {
				throw (IOException)t;
			}
			else {
				throw new RuntimeException(t);
			}
		}

	}

	@Override
	public boolean reachedEnd() throws IOException {

		try {
			return !sinkStage.hasNextTuple() && !this.sourceStage.readNextRecord();
		}
		catch( OutOfMemoryError error ) {
			throw error;
		}
		catch( IOException exception ) {
			throw exception;
		}
		catch( Throwable throwable ) {
			if( throwable instanceof CascadingException ) {
				throw (CascadingException) throwable;
			}

			throw new FlowException( "internal error during TapInputFormat execution", throwable );
		}
	}

	@Override
	public Tuple nextRecord(Tuple record) throws IOException {

		if(this.reachedEnd()) {
			return null;
		}
		else {
			return sinkStage.fetchNextTuple();
		}
	}

	@Override
	public void close() throws IOException {
		try {
			streamGraph.cleanup();
		}
		finally {

			long processEndTime = System.currentTimeMillis();
			flowProcess.increment(SliceCounters.Process_End_Time, processEndTime);
			flowProcess.increment( SliceCounters.Process_Duration, processEndTime - this.processBeginTime );

			String message = "flow node id: " + flowNode.getID();
			logMemory( LOG, message + ", mem on close" );
			logCounters( LOG, message + ", counter:", flowProcess );
		}
	}


	// --------------------------------------------------------------------------------------------
	//  Helper methods
	// --------------------------------------------------------------------------------------------

	@Override
	public BaseStatistics getStatistics(BaseStatistics cachedStats) throws IOException {
		// only gather base statistics for FileInputFormats
		if (!(mapredInputFormat instanceof FileInputFormat)) {
			return null;
		}

		final FileBaseStatistics cachedFileStats = (cachedStats != null && cachedStats instanceof FileBaseStatistics) ?
				(FileBaseStatistics) cachedStats : null;

		try {
			final org.apache.hadoop.fs.Path[] paths = FileInputFormat.getInputPaths(this.jobConf);

			return getFileStats(cachedFileStats, paths, new ArrayList<FileStatus>(1));
		} catch (IOException ioex) {
			if (LOG.isWarnEnabled()) {
				LOG.warn("Could not determine statistics due to an io error: "
						+ ioex.getMessage());
			}
		} catch (Throwable t) {
			if (LOG.isErrorEnabled()) {
				LOG.error("Unexpected problem while getting the file statistics: "
						+ t.getMessage(), t);
			}
		}

		// no statistics available
		return null;
	}

	@Override
	public HadoopInputSplit[] createInputSplits(int minNumSplits)
			throws IOException {

		org.apache.hadoop.mapred.InputSplit[] splitArray = mapredInputFormat.getSplits(jobConf, minNumSplits);
		HadoopInputSplit[] hiSplit = new HadoopInputSplit[splitArray.length];
		for (int i = 0; i < splitArray.length; i++) {
			hiSplit[i] = new HadoopInputSplit(i, splitArray[i], jobConf);
		}
		return hiSplit;
	}

	@Override
	public InputSplitAssigner getInputSplitAssigner(HadoopInputSplit[] inputSplits) {
		return new LocatableInputSplitAssigner(inputSplits);
	}

	private FileBaseStatistics getFileStats(FileBaseStatistics cachedStats, org.apache.hadoop.fs.Path[] hadoopFilePaths,
											ArrayList<FileStatus> files) throws IOException {

		long latestModTime = 0L;

		// get the file info and check whether the cached statistics are still valid.
		for (org.apache.hadoop.fs.Path hadoopPath : hadoopFilePaths) {

			final Path filePath = new Path(hadoopPath.toUri());
			final FileSystem fs = FileSystem.get(filePath.toUri());

			final FileStatus file = fs.getFileStatus(filePath);
			latestModTime = Math.max(latestModTime, file.getModificationTime());

			// enumerate all files and check their modification time stamp.
			if (file.isDir()) {
				FileStatus[] fss = fs.listStatus(filePath);
				files.ensureCapacity(files.size() + fss.length);

				for (FileStatus s : fss) {
					if (!s.isDir()) {
						files.add(s);
						latestModTime = Math.max(s.getModificationTime(), latestModTime);
					}
				}
			} else {
				files.add(file);
			}
		}

		// check whether the cached statistics are still valid, if we have any
		if (cachedStats != null && latestModTime <= cachedStats.getLastModificationTime()) {
			return cachedStats;
		}

		// calculate the whole length
		long len = 0;
		for (FileStatus s : files) {
			len += s.getLen();
		}

		// sanity check
		if (len <= 0) {
			len = BaseStatistics.SIZE_UNKNOWN;
		}

		return new FileBaseStatistics(latestModTime, len, BaseStatistics.AVG_RECORD_BYTES_UNKNOWN);
	}

} public class IdMapper implements MapFunction<Tuple, Tuple> {

	@Override
	public Tuple map(Tuple value) throws Exception {
		return value;
	}
} public class TupleTypeInfo extends CompositeType<Tuple> {

	private final static int NEG_FIELD_POS_OFFSET = Integer.MAX_VALUE / 2;

	private Fields schema;

	private final int length;
	private LinkedHashMap<String, FieldTypeInfo> fieldTypes;
	private HashMap<String, Integer> fieldIndexes;

	public TupleTypeInfo(Fields schema) {
		super(Tuple.class);

		this.schema = schema;
		this.fieldIndexes = new HashMap<String, Integer>();

		if(schema.isDefined()) {
			this.length = schema.size();
			this.fieldTypes = new LinkedHashMap<String, FieldTypeInfo>(this.length);

			Comparator[] comps = schema.getComparators();
			Class[] typeClasses = schema.getTypesClasses();

			for(int i=0; i<length; i++) {
				String fieldName = getFieldName(i);
				FieldTypeInfo fieldType = getFieldTypeInfo(i, typeClasses, comps);

				this.fieldTypes.put(fieldName, fieldType);
				this.fieldIndexes.put(fieldName, i);
			}
		}
		else {
			if(schema.isUnknown()) {
				this.length = -1;
				this.fieldTypes = new LinkedHashMap<String, FieldTypeInfo>(16);

				this.fieldTypes.put("0", new FieldTypeInfo());
				this.fieldIndexes.put("0", 0);
			}
			else {
				throw new IllegalArgumentException("Unsupported Fields: "+schema);
			}
		}
	}

	public String[] registerKeyFields(Fields keyFields) {

		int[] keyPos;
		if(keyFields.isAll()) {
			keyPos = new int[this.schema.size()];
			for(int i=0; i<keyPos.length; i++) {
				keyPos[i] = i;
			}
		}
		else {
			keyPos = this.schema.getPos(keyFields);
		}

		String[] serdePos = new String[keyPos.length];

		Comparator[] keyComps = keyFields.getComparators();
		Class[] keyTypes = keyFields.getTypesClasses();

		for(int j=0; j<keyPos.length; j++) {

			String fieldName = getFieldName(keyPos[j]);

			if(!this.fieldIndexes.containsKey(fieldName)) {
				this.fieldIndexes.put(fieldName, keyPos[j]);
			}

			FieldTypeInfo fieldType = this.fieldTypes.get(fieldName);
			if(fieldType == null) {
				fieldType = new FieldTypeInfo();
				this.fieldTypes.put(fieldName, fieldType);
			}

			// set serde position
			serdePos[j] = fieldName;

			if(keyTypes != null && keyTypes.length > j && keyTypes[j] != null) {
				// set key type
				fieldType.setFieldType(keyTypes[j]);
			}

			if(keyComps != null && keyComps.length > j && keyComps[j] != null) {
				// set custom key comparator
				fieldType.setCustomComparator(keyComps[j]);
			}
		}

		return serdePos;
	}

	public Fields getSchema() {
		return this.schema;
	}

	@Override
	public boolean isBasicType() {
		return false;
	}

	@Override
	public boolean isTupleType() {
		return false;
	}

	@Override
	public int getArity() {
		if(this.length > 0) {
			return this.length;
		}
		else {
			return 1;
		}
	}

	@Override
	public int getTotalFields() {
		return this.getArity();
	}

	@Override
	public Class<Tuple> getTypeClass() {
		return Tuple.class;
	}

	@Override
	public String[] getFieldNames() {
		return (String[])this.fieldTypes.keySet().toArray();
	}

	@Override
	public int getFieldIndex(String fieldName) {

		try {
			int idx = this.fieldIndexes.get(fieldName);
			if(this.fieldTypes.get(fieldName) != null) {
				return getFlinkPos(idx);
			}
			else {
				throw new IndexOutOfBoundsException("Field index out of bounds.");
			}
		}
		catch(NumberFormatException nfe) {
			throw new IllegalArgumentException("Only numeric keys supported for Cascading tuples");
		}
	}

	@Override
	public <X> TypeInformation<X> getTypeAt(String fieldExpression) {
		int idx = getFieldIndex(fieldExpression);
		return this.getTypeAt(idx);
	}

	@Override
	public <X> TypeInformation<X> getTypeAt(int idx) {

		idx = getCascadingPos(idx);

		String fieldName = getFieldName(idx);
		if(this.fieldTypes.get(fieldName) != null) {
			return (TypeInformation<X>) this.fieldTypes.get(fieldName);
		}
		else {
			throw new IndexOutOfBoundsException("Field index out of bounds.");
		}
	}

	@Override
	protected TypeComparatorBuilder<Tuple> createTypeComparatorBuilder() {
		// not required
		throw new UnsupportedOperationException();
	}

	@Override
	public void getFlatFields(String fieldExpression, int offset, List<FlatFieldDescriptor> list) {

		if(fieldExpression.equals("*")) {
			for(int i=0; i<this.length; i++) {
				list.add(new FlatFieldDescriptor(offset+i, getTypeAt(i)));
			}
		}
		else {
			int fieldIdx = this.getFieldIndex(fieldExpression);
			list.add(new FlatFieldDescriptor(offset+fieldIdx, getTypeAt(fieldIdx)));
		}
	}

	@Override
	public TypeSerializer<Tuple> createSerializer(ExecutionConfig config) {

		if(this.length > 0) {
			// create serializer for tuple with schema

			TypeSerializer[] fieldSers;
			fieldSers = new TypeSerializer[this.length];
			for(String field : this.fieldTypes.keySet()) {
				Integer fieldIdx = Integer.parseInt(field);
				fieldSers[fieldIdx] = this.fieldTypes.get(field).createSerializer(config);
			}
			return new DefinedTupleSerializer(this.schema, fieldSers);
		}
		else {
			// create serializer for tuple without schema

			TypeSerializer defaultFieldSer = new FieldTypeInfo().createSerializer(config);
			return new UnknownTupleSerializer(defaultFieldSer);
		}
	}

	@Override
	public TypeComparator<Tuple> createComparator(int[] keyIdxs, boolean[] orders, int offset, ExecutionConfig config) {

		if(keyIdxs.length == 0) {
			throw new RuntimeException("Empty key indexes");
		}
		if(offset != 0) {
			throw new RuntimeException("Only 0 offset supported.");
		}

		// get key comparators
		TypeComparator<?>[] keyComps = new TypeComparator[keyIdxs.length];
		for(int i = 0; i < keyIdxs.length; i++) {
			keyComps[i] = ((AtomicType)this.getTypeAt(keyIdxs[i])).createComparator(orders[i], config);
		}

		if(length > 0) {
			// comparator for tuples with defined schema
			int maxKey = 0;
			for(int i = 0; i < keyIdxs.length; ++i) {
				int key = keyIdxs[i];
				maxKey = Math.max(maxKey, key);
			}

			// get field serializers up to max key
			TypeSerializer[] serializers = new TypeSerializer[maxKey + 1];
			for(int i = 0; i <= maxKey; ++i) {
				serializers[i] = this.fieldTypes.get(Integer.toString(i)).createSerializer(config);
			}

			return new DefinedTupleComparator(keyIdxs, keyComps, serializers, this.length);
		}
		else {
			// comparator for unknown tuples
			int[] cascadingKeyIdx = new int[keyIdxs.length];
			for(int i=0; i<cascadingKeyIdx.length; i++) {
				cascadingKeyIdx[i] = getCascadingPos(keyIdxs[i]);
			}

			return new UnknownTupleComparator(cascadingKeyIdx, keyComps, new FieldTypeInfo().createSerializer(config));
		}

	}

	@Override
	public boolean equals(Object o) {

		if (o instanceof TupleTypeInfo) {
			return this.schema.equalsFields(((TupleTypeInfo) o).getSchema());
		}
		else {
			return false;
		}
	}

	private FieldTypeInfo getFieldTypeInfo(int pos, Class[] typeClasses, Comparator[] fieldComparators) {
		if(typeClasses != null && pos >= typeClasses.length) {
			throw new ArrayIndexOutOfBoundsException("Fields position out of bounds");
		}
		if(fieldComparators != null && pos >= fieldComparators.length) {
			throw new ArrayIndexOutOfBoundsException("Fields position out of bounds");
		}

		FieldTypeInfo fieldTypeInfo;
		if(typeClasses != null && typeClasses[pos] != null) {
			fieldTypeInfo = new FieldTypeInfo(typeClasses[pos]);
		}
		else {
			fieldTypeInfo = new FieldTypeInfo();
		}

		// set custom field comparator if any
		if(fieldComparators != null && fieldComparators[pos] != null) {
			fieldTypeInfo.setCustomComparator(fieldComparators[pos]);
		}

		return fieldTypeInfo;
	}

	// this is an ugly hack to support negative relative positions on unknown schemas.
	private int getFlinkPos(int cascadingPos) {
		if(cascadingPos >= 0) {
			if(cascadingPos > NEG_FIELD_POS_OFFSET) {
				throw new RuntimeException("Maximum key position "+NEG_FIELD_POS_OFFSET+" exceeded");
			}

			return cascadingPos;
		}
		else {
			return NEG_FIELD_POS_OFFSET + (-1 * cascadingPos);
		}
	}

	// this is an ugly hack to support negative relative positions on unknown schemas.
	private int getCascadingPos(int flinkPos) {
		if(flinkPos > NEG_FIELD_POS_OFFSET) {
			return (flinkPos - NEG_FIELD_POS_OFFSET) * -1;
		}
		else {
			return flinkPos;
		}
	}

	private String getFieldName(int index) {
		if(index >= 0) {
			return Integer.toString(index);
		}
		else {
			return "neg"+(-1*index);
		}
	}
} public class TupleArrayTypeInfo extends TypeInformation<Tuple[]> {

	private final int length;
	private final int fillLength;
	private final TupleTypeInfo[] tupleTypes;

	public TupleArrayTypeInfo(int length, Fields[] fields) {
		this.length = length;
		this.fillLength = fields.length;

		this.tupleTypes = new TupleTypeInfo[this.fillLength];
		for(int i=0; i<this.fillLength; i++) {
			this.tupleTypes[i] = new TupleTypeInfo(fields[i]);
		}
	}

	@Override
	public boolean isBasicType() {
		return false;
	}

	@Override
	public boolean isTupleType() {
		return false;
	}

	@Override
	public int getArity() {
		return 1;
	}

	@Override
	public int getTotalFields() {
		return 1;
	}

	@Override
	public Class<Tuple[]> getTypeClass() {
		return Tuple[].class;
	}

	@Override
	public boolean isKeyType() {
		return false;
	}

	@Override
	public TypeSerializer<Tuple[]> createSerializer(ExecutionConfig executionConfig) {

		TypeSerializer<Tuple>[] tupleSerializers = new TypeSerializer[this.fillLength];
		for(int i=0; i<this.fillLength; i++) {
			tupleSerializers[i] = this.tupleTypes[i].createSerializer(executionConfig);
		}

		return new TupleArraySerializer(this.length, tupleSerializers);
	}

	@Override
	public String toString() {
		return "Tuple["+this.length+"] "+ Arrays.toString(tupleTypes);
	}

	@Override
	public boolean equals(Object obj) {
		if (obj instanceof TupleArrayTypeInfo) {
			TupleArrayTypeInfo other = (TupleArrayTypeInfo) obj;

			// sloppy equals check, Cascading does not always provide types.
			// We rely on Cascading to do the type checking here.
			return other.canEqual(this) &&
					this.length == other.length &&
					Arrays.equals(this.tupleTypes, other.tupleTypes);
		}
		else {
			return false;
		}
	}

	@Override
	public int hashCode() {
		return 31 * this.length + Arrays.hashCode(this.tupleTypes);
	}

	@Override
	public boolean canEqual(Object obj) {
		return obj instanceof TupleArrayTypeInfo;
	}
} public class FlinkConfigConverter {

	public static Configuration toHadoopConfig(org.apache.flink.configuration.Configuration flinkConfig) {
		if(flinkConfig == null) {
			return null;
		}
		Configuration hadoopConfig = new Configuration();
		for(String key : flinkConfig.keySet()) {
			hadoopConfig.set(key, flinkConfig.getString(key, null));
		}
		return hadoopConfig;
	}

	public static org.apache.flink.configuration.Configuration toFlinkConfig(Configuration hadoopConfig) {
		if(hadoopConfig == null) {
			return null;
		}
		org.apache.flink.configuration.Configuration flinkConfig = new org.apache.flink.configuration.Configuration();
		for(Map.Entry<String, String> e : hadoopConfig) {
			flinkConfig.setString(e.getKey(), e.getValue());
		}
		return flinkConfig;
	}

}
import cascading.flow.FlowElement;
import cascading.flow.FlowException;
import cascading.flow.FlowNode;
import cascading.flow.FlowProcess;
import cascading.flow.hadoop.util.HadoopUtil;
import cascading.flow.planner.BaseFlowStep;
import cascading.flow.planner.FlowStepJob;
import cascading.flow.planner.Scope;
import cascading.flow.planner.graph.ElementGraph;
import cascading.flow.planner.graph.Extent;
import cascading.flow.planner.process.FlowNodeGraph;
import cascading.management.state.ClientState;
import cascading.pipe.Boundary;
import cascading.pipe.CoGroup;
import cascading.pipe.GroupBy;
import cascading.pipe.HashJoin;
import cascading.pipe.Merge;
import cascading.pipe.Pipe;
import cascading.pipe.Splice;
import cascading.pipe.joiner.BufferJoin;
import cascading.pipe.joiner.InnerJoin;
import cascading.pipe.joiner.Joiner;
import cascading.pipe.joiner.LeftJoin;
import cascading.property.ConfigDef;
import cascading.tap.Tap;
import cascading.tap.hadoop.io.MultiInputFormat;
import cascading.tuple.Fields;
import cascading.tuple.Tuple;
import com.dataartisans.flink.cascading.runtime.coGroup.bufferJoin.BufferJoinKeyExtractor;
import com.dataartisans.flink.cascading.runtime.coGroup.bufferJoin.CoGroupBufferReducer;
import com.dataartisans.flink.cascading.runtime.coGroup.regularJoin.CoGroupReducer;
import com.dataartisans.flink.cascading.runtime.coGroup.regularJoin.TupleAppendOuterJoiner;
import com.dataartisans.flink.cascading.runtime.coGroup.regularJoin.TupleOuterJoiner;
import com.dataartisans.flink.cascading.runtime.groupBy.GroupByReducer;
import com.dataartisans.flink.cascading.runtime.hashJoin.NaryHashJoinJoiner;
import com.dataartisans.flink.cascading.runtime.util.FlinkFlowProcess;
import com.dataartisans.flink.cascading.runtime.hashJoin.BinaryHashJoinJoiner;
import com.dataartisans.flink.cascading.runtime.hashJoin.JoinPrepareMapper;
import com.dataartisans.flink.cascading.runtime.hashJoin.TupleAppendCrosser;
import com.dataartisans.flink.cascading.runtime.hashJoin.TupleAppendJoiner;
import com.dataartisans.flink.cascading.runtime.hashJoin.HashJoinMapper;
import com.dataartisans.flink.cascading.runtime.each.EachMapper;
import com.dataartisans.flink.cascading.runtime.sink.TapOutputFormat;
import com.dataartisans.flink.cascading.runtime.source.TapInputFormat;
import com.dataartisans.flink.cascading.runtime.util.IdMapper;
import com.dataartisans.flink.cascading.types.tuple.TupleTypeInfo;
import com.dataartisans.flink.cascading.types.tuplearray.TupleArrayTypeInfo;
import com.dataartisans.flink.cascading.util.FlinkConfigConverter;
import org.apache.flink.api.common.Plan;
import org.apache.flink.api.common.operators.Order;
import org.apache.flink.api.common.operators.base.JoinOperatorBase.JoinHint;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.GroupReduceOperator;
import org.apache.flink.api.java.operators.JoinOperator;
import org.apache.flink.api.java.operators.Operator;
import org.apache.flink.api.java.operators.PartitionOperator;
import org.apache.flink.api.java.operators.SortPartitionOperator;
import org.apache.flink.api.java.operators.SortedGrouping;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapred.JobConf;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
		Joiner joiner = hashJoin.getJoiner();

		// check if joiner is a Scalding WrappedJoiner and
		//   try to extract the joiner which is wrapped inside
		if (joiner.getClass().getName().equals("com.twitter.scalding.WrappedJoiner")) {
			try {
				Field joinerField = joiner.getClass().getDeclaredField("joiner");
				joinerField.setAccessible(true);
				joiner = (Joiner)joinerField.get(joiner);
			}
			catch(NoSuchFieldException | IllegalAccessException nsfe) {
				nsfe.printStackTrace();
				LOG.warn("Could not extract joiner from Scalding's WrappedJoiner. " +
						"Will continue without extracting joiner.");
			}
		}

		int numJoinInputs = hashJoin.isSelfJoin() ? hashJoin.getNumSelfJoins() + 1 : inputs.size();

		Fields[] inputFields = new Fields[numJoinInputs];
		Fields[] keyFields = new Fields[numJoinInputs];
		String[][] flinkKeys = new String[numJoinInputs][];

		List<DataSet<Tuple>> joinInputs = computeSpliceInputsFieldsKeys(hashJoin, node, inputs, inputFields, keyFields, flinkKeys);

		if(keyFields[0].isNone()) {
			// Cartesian product
			return translateInnerCrossProduct(node, joinInputs);
		}
		else if(joiner.getClass().equals(InnerJoin.class)) {
			// inner join with keys
			return translateInnerHashJoin(node, joinInputs, inputFields, keyFields, flinkKeys);
		}
		else if (joiner.getClass().equals(LeftJoin.class)) {
			return translateLeftHashJoin(node, joinInputs, inputFields, keyFields, flinkKeys);
		}
		else {
			System.out.println(joiner.getClass().getName());
			throw new FlowException("HashJoin does only support InnerJoin and LeftJoin but is " +
					joiner.getClass().getName());
		}
	}

	private DataSet<Tuple> translateInnerHashJoin(FlowNode node, List<DataSet<Tuple>> inputs, Fields[] inputFields, Fields[] keyFields, String[][] flinkKeys) {

		int numJoinInputs = inputs.size();

		// get out fields of node
		Scope outScope = getOutScope(node);
		Fields outFields;
		if (outScope.isEvery()) {
			outFields = outScope.getOutGroupingFields();
		} else {
			outFields = outScope.getOutValuesFields();
		}
		registerKryoTypes(outFields);

		int probeSideDOP = ((Operator)inputs.get(0)).getParallelism();

		if(numJoinInputs == 2) {
			// binary join

			return inputs.get(0).join(inputs.get(1), JoinHint.BROADCAST_HASH_SECOND)
					.where(flinkKeys[0]).equalTo(flinkKeys[1])
					.with(new BinaryHashJoinJoiner(node, inputFields[0], keyFields[0]))
					.withParameters(this.getFlinkNodeConfig(node))
					.setParallelism(probeSideDOP)
					.returns(new TupleTypeInfo(outFields))
					.name("hashjoin-" + node.getID());

		}
		else {
			// nary join

			TupleTypeInfo keysTypeInfo = inputFields[0].isDefined() ?
					new TupleTypeInfo(inputFields[0].select(keyFields[0])) :
					new TupleTypeInfo(Fields.UNKNOWN);
			keysTypeInfo.registerKeyFields(keyFields[0]);


			TypeInformation<Tuple2<Tuple, Tuple[]>> tupleJoinListsTypeInfo =
					new org.apache.flink.api.java.typeutils.TupleTypeInfo<>(
							keysTypeInfo,
							new TupleArrayTypeInfo(numJoinInputs-1, Arrays.copyOf(inputFields, 1))
					);

			int mapDop = ((Operator) inputs.get(0)).getParallelism();

			// prepare tuple list for join
			DataSet<Tuple2<Tuple, Tuple[]>> tupleJoinLists = inputs.get(0)
					.map(new JoinPrepareMapper(numJoinInputs - 1, inputFields[0], keyFields[0]))
					.returns(tupleJoinListsTypeInfo)
					.setParallelism(mapDop)
					.name("hashjoin-" + node.getID());

			for (int i = 0; i < flinkKeys[0].length; i++) {
				flinkKeys[0][i] = "f0." + i;
			}

			// join all inputs except last
			for (int i = 1; i < inputs.size()-1; i++) {

				tupleJoinListsTypeInfo =
						new org.apache.flink.api.java.typeutils.TupleTypeInfo<>(
								keysTypeInfo,
								new TupleArrayTypeInfo(numJoinInputs-1, Arrays.copyOf(inputFields, i+1))
						);

				tupleJoinLists = tupleJoinLists.join(inputs.get(i), JoinHint.BROADCAST_HASH_SECOND)
						.where(flinkKeys[0]).equalTo(flinkKeys[i])
						.with(new TupleAppendJoiner(i))
						.returns(tupleJoinListsTypeInfo)
						.withForwardedFieldsFirst(flinkKeys[0])
						.setParallelism(probeSideDOP)
						.name("hashjoin-" + node.getID());
			}

			// join last input
			return tupleJoinLists.join(inputs.get(numJoinInputs-1), JoinHint.BROADCAST_HASH_SECOND)
					.where(flinkKeys[0]).equalTo(flinkKeys[numJoinInputs-1])
					.with(new NaryHashJoinJoiner(node, numJoinInputs))