//TODO ??? need to create a new TezConfig ???
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;
	}
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);
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();
// 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;
}