private int translateChannel(Channel input, int inputIndex, FlinkVertex targetVertex,
								TezTaskConfig targetVertexConfig, boolean isBroadcast) throws Exception
	{
		final PlanNode inputPlanNode = input.getSource();
		final Iterator<Channel> allInChannels;
		allInChannels = Collections.singletonList(input).iterator();
		// check that the type serializer is consistent
		TypeSerializerFactory<?> typeSerFact = null;
		while (allInChannels.hasNext()) {
			final Channel inConn = allInChannels.next();
			if (typeSerFact == null) {
				typeSerFact = inConn.getSerializer();
			} else if (!typeSerFact.equals(inConn.getSerializer())) {
				throw new CompilerException("Conflicting types in union operator.");
			}
			final PlanNode sourceNode = inConn.getSource();
			FlinkVertex sourceVertex = this.vertices.get(sourceNode);
			TezTaskConfig sourceVertexConfig = sourceVertex.getConfig(); //TODO ??? need to create a new TezConfig ???
			connectJobVertices(
					inConn, inputIndex, sourceVertex, sourceVertexConfig, targetVertex, targetVertexConfig, isBroadcast);
		}
		// the local strategy is added only once. in non-union case that is the actual edge,
		// in the union case, it is the edge between union and the target node
		addLocalInfoFromChannelToConfig(input, targetVertexConfig, inputIndex, isBroadcast);
		return 1;
	}
private int translateChannel(Channel input, int inputIndex, FlinkVertex targetVertex,
								TezTaskConfig targetVertexConfig, boolean isBroadcast) throws Exception
	{
		final PlanNode inputPlanNode = input.getSource();
		final Iterator<Channel> allInChannels;
		allInChannels = Collections.singletonList(input).iterator();
	
		TypeSerializerFactory<?> typeSerFact = null;
		while (allInChannels.hasNext()) {
			final Channel inConn = allInChannels.next();
			if (typeSerFact == null) {
				typeSerFact = inConn.getSerializer();
			} else if (!typeSerFact.equals(inConn.getSerializer())) {
				throw new CompilerException("Conflicting types in union operator.");
			}
			final PlanNode sourceNode = inConn.getSource();
			FlinkVertex sourceVertex = this.vertices.get(sourceNode);
			TezTaskConfig sourceVertexConfig = sourceVertex.getConfig();
			connectJobVertices(
					inConn, inputIndex, sourceVertex, sourceVertexConfig, targetVertex, targetVertexConfig, isBroadcast);
		}
	
	
		addLocalInfoFromChannelToConfig(input, targetVertexConfig, inputIndex, isBroadcast);
		return 1;
	}
private int translatechannel(channel input, int inputindex, flinkvertex targetvertex, teztaskconfig targetvertexconfig, boolean isbroadcast) throws exception { final plannode inputplannode = input.getsource(); final iterator<channel> allinchannels; allinchannels = collections.singletonlist(input).iterator(); typeserializerfactory<?> typeserfact = null; while (allinchannels.hasnext()) { final channel inconn = allinchannels.next(); if (typeserfact == null) { typeserfact = inconn.getserializer(); } else if (!typeserfact.equals(inconn.getserializer())) { throw new compilerexception("conflicting types in union operator."); } final plannode sourcenode = inconn.getsource(); flinkvertex sourcevertex = this.vertices.get(sourcenode); teztaskconfig sourcevertexconfig = sourcevertex.getconfig(); connectjobvertices( inconn, inputindex, sourcevertex, sourcevertexconfig, targetvertex, targetvertexconfig, isbroadcast); } addlocalinfofromchanneltoconfig(input, targetvertexconfig, inputindex, isbroadcast); return 1; }