[hotfix] [optimizer] Normalize job plan operator formatting

When printing the job plan the operator description is typically
formatted as the operator name followed by the user given or generated
function name in parenthesis. For example, "Reduce (My Function)".

This normalizes the node names to include a space between the operator
and function names.

This closes #4383

flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/SolutionSetNode.java

@@ -46,7 +46,7 @@ public class SolutionSetNode extends AbstractPartialSolutionNode {
 	// --------------------------------------------------------------------------------------------
 	
 	public void setCandidateProperties(GlobalProperties gProps, LocalProperties lProps, Channel initialInput) {
-		this.cachedPlans = Collections.<PlanNode>singletonList(new SolutionSetPlanNode(this, "SolutionSet("+this.getOperator().getName()+")", gProps, lProps, initialInput));
+		this.cachedPlans = Collections.<PlanNode>singletonList(new SolutionSetPlanNode(this, "SolutionSet ("+this.getOperator().getName()+")", gProps, lProps, initialInput));
 	}
 	
 	public SolutionSetPlanNode getCurrentSolutionSetPlanNode() {

flink-optimizer/src/main/java/org/apache/flink/optimizer/operators/AbstractSortMergeJoinDescriptor.java

@@ -67,7 +67,7 @@ public abstract class AbstractSortMergeJoinDescriptor extends AbstractJoinDescri
 			inputOrders = tmp;
 		}
 
-		String nodeName = String.format("%s(%s)", getNodeName(), node.getOperator().getName());
+		String nodeName = String.format("%s (%s)", getNodeName(), node.getOperator().getName());
 		return new DualInputPlanNode(node, nodeName, in1, in2, getStrategy(), this.keys1, this.keys2, inputOrders);
 	}
 

flink-optimizer/src/main/java/org/apache/flink/optimizer/operators/CartesianProductDescriptor.java

@@ -93,7 +93,7 @@ public abstract class CartesianProductDescriptor extends OperatorDescriptorDual
 	
 	@Override
 	public DualInputPlanNode instantiate(Channel in1, Channel in2, TwoInputNode node) {
-		return new DualInputPlanNode(node, "Cross("+node.getOperator().getName()+")", in1, in2, getStrategy());
+		return new DualInputPlanNode(node, "Cross ("+node.getOperator().getName()+")", in1, in2, getStrategy());
 	}
 	
 	@Override

flink-optimizer/src/main/java/org/apache/flink/optimizer/operators/GroupReduceWithCombineProperties.java

@@ -116,7 +116,7 @@ public final class GroupReduceWithCombineProperties extends OperatorDescriptorSi
 			GroupReduceNode combinerNode = ((GroupReduceNode) node).getCombinerUtilityNode();
 			combinerNode.setParallelism(in.getSource().getParallelism());
 
-			SingleInputPlanNode combiner = new SingleInputPlanNode(combinerNode, "Combine("+node.getOperator()
+			SingleInputPlanNode combiner = new SingleInputPlanNode(combinerNode, "Combine ("+node.getOperator()
 					.getName()+")", toCombiner, DriverStrategy.SORTED_GROUP_COMBINE);
 			combiner.setCosts(new Costs(0, 0));
 			combiner.initProperties(toCombiner.getGlobalProperties(), toCombiner.getLocalProperties());

flink-optimizer/src/main/java/org/apache/flink/optimizer/operators/HashFullOuterJoinBuildFirstDescriptor.java

@@ -54,7 +54,7 @@ public class HashFullOuterJoinBuildFirstDescriptor extends AbstractJoinDescripto
 	@Override
 	public DualInputPlanNode instantiate(Channel in1, Channel in2, TwoInputNode node) {
 
-		String nodeName = "FullOuterJoin("+node.getOperator().getName()+")";
+		String nodeName = "FullOuterJoin ("+node.getOperator().getName()+")";
 		return new DualInputPlanNode(node, nodeName, in1, in2, getStrategy(), this.keys1, this.keys2);
 	}
 

flink-optimizer/src/main/java/org/apache/flink/optimizer/operators/HashFullOuterJoinBuildSecondDescriptor.java

@@ -53,7 +53,7 @@ public class HashFullOuterJoinBuildSecondDescriptor extends AbstractJoinDescript
 	@Override
 	public DualInputPlanNode instantiate(Channel in1, Channel in2, TwoInputNode node) {
 
-		String nodeName = "FullOuterJoin("+node.getOperator().getName()+")";
+		String nodeName = "FullOuterJoin ("+node.getOperator().getName()+")";
 		return new DualInputPlanNode(node, nodeName, in1, in2, getStrategy(), this.keys1, this.keys2);
 	}
 

flink-optimizer/src/main/java/org/apache/flink/optimizer/operators/HashJoinBuildFirstProperties.java

@@ -79,7 +79,7 @@ public class HashJoinBuildFirstProperties extends AbstractJoinDescriptor {
 		else {
 			strategy = DriverStrategy.HYBRIDHASH_BUILD_FIRST;
 		}
-		return new DualInputPlanNode(node, "Join("+node.getOperator().getName()+")", in1, in2, strategy, this.keys1, this.keys2);
+		return new DualInputPlanNode(node, "Join ("+node.getOperator().getName()+")", in1, in2, strategy, this.keys1, this.keys2);
 	}
 	
 	@Override

flink-optimizer/src/main/java/org/apache/flink/optimizer/operators/HashLeftOuterJoinBuildFirstDescriptor.java

@@ -54,7 +54,7 @@ public class HashLeftOuterJoinBuildFirstDescriptor extends AbstractJoinDescripto
 	@Override
 	public DualInputPlanNode instantiate(Channel in1, Channel in2, TwoInputNode node) {
 
-		String nodeName = "LeftOuterJoin("+node.getOperator().getName()+")";
+		String nodeName = "LeftOuterJoin ("+node.getOperator().getName()+")";
 		return new DualInputPlanNode(node, nodeName, in1, in2, getStrategy(), this.keys1, this.keys2);
 	}
 

flink-optimizer/src/main/java/org/apache/flink/optimizer/operators/HashLeftOuterJoinBuildSecondDescriptor.java

@@ -57,7 +57,7 @@ public class HashLeftOuterJoinBuildSecondDescriptor extends AbstractJoinDescript
 	@Override
 	public DualInputPlanNode instantiate(Channel in1, Channel in2, TwoInputNode node) {
 
-		String nodeName = "LeftOuterJoin("+node.getOperator().getName()+")";
+		String nodeName = "LeftOuterJoin ("+node.getOperator().getName()+")";
 		return new DualInputPlanNode(node, nodeName, in1, in2, getStrategy(), this.keys1, this.keys2);
 	}
 	

flink-optimizer/src/main/java/org/apache/flink/optimizer/operators/HashRightOuterJoinBuildFirstDescriptor.java

@@ -57,7 +57,7 @@ public class HashRightOuterJoinBuildFirstDescriptor extends AbstractJoinDescript
 	@Override
 	public DualInputPlanNode instantiate(Channel in1, Channel in2, TwoInputNode node) {
 
-		String nodeName = "RightOuterJoin("+node.getOperator().getName()+")";
+		String nodeName = "RightOuterJoin ("+node.getOperator().getName()+")";
 		return new DualInputPlanNode(node, nodeName, in1, in2, getStrategy(), this.keys1, this.keys2);
 	}
 	

flink-optimizer/src/main/java/org/apache/flink/optimizer/operators/HashRightOuterJoinBuildSecondDescriptor.java

@@ -54,7 +54,7 @@ public class HashRightOuterJoinBuildSecondDescriptor extends AbstractJoinDescrip
 	@Override
 	public DualInputPlanNode instantiate(Channel in1, Channel in2, TwoInputNode node) {
 
-		String nodeName = "RightOuterJoin("+node.getOperator().getName()+")";
+		String nodeName = "RightOuterJoin ("+node.getOperator().getName()+")";
 		return new DualInputPlanNode(node, nodeName, in1, in2, getStrategy(), this.keys1, this.keys2);
 	}
 

flink-optimizer/src/main/java/org/apache/flink/optimizer/operators/PartialGroupProperties.java

@@ -51,7 +51,7 @@ public final class PartialGroupProperties extends OperatorDescriptorSingle {
 		GroupReduceNode combinerNode = new GroupReduceNode((GroupReduceOperatorBase<?, ?, ?>) node.getOperator());
 		combinerNode.setParallelism(in.getSource().getParallelism());
 
-		SingleInputPlanNode combiner = new SingleInputPlanNode(combinerNode, "Combine("+node.getOperator().getName()+")", in,
+		SingleInputPlanNode combiner = new SingleInputPlanNode(combinerNode, "Combine ("+node.getOperator().getName()+")", in,
 				DriverStrategy.SORTED_GROUP_COMBINE);
 		// sorting key info
 		combiner.setDriverKeyInfo(in.getLocalStrategyKeys(), in.getLocalStrategySortOrder(), 0);

flink-optimizer/src/main/java/org/apache/flink/optimizer/plantranslate/JobGraphGenerator.java

@@ -1283,7 +1283,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
 		headConfig.setRelativeBackChannelMemory(relativeMemForBackChannel);
 		
 		// --------------------------- create the sync task ---------------------------
-		final JobVertex sync = new JobVertex("Sync(" + bulkNode.getNodeName() + ")");
+		final JobVertex sync = new JobVertex("Sync (" + bulkNode.getNodeName() + ")");
 		sync.setResources(bulkNode.getMinResources(), bulkNode.getPreferredResources());
 		sync.setInvokableClass(IterationSynchronizationSinkTask.class);
 		sync.setParallelism(1);