Repository: flink
Updated Branches:
  refs/heads/release-1.1 c5b391c5e -> a7f6594b6


[FLINK-2662] [dataSet] [optimizer] Fix merging of unions with multiple outputs.

Translate union with N outputs into N unions with single output.


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/a7f6594b
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/a7f6594b
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/a7f6594b

Branch: refs/heads/release-1.1
Commit: a7f6594b6b47b91242cbb0a13ea4efc5508adcfc
Parents: c5b391c
Author: Fabian Hueske <fhue...@apache.org>
Authored: Fri Sep 16 18:40:32 2016 +0200
Committer: Fabian Hueske <fhue...@apache.org>
Committed: Tue Sep 20 21:55:52 2016 +0200

----------------------------------------------------------------------
 .../api/java/operators/OperatorTranslation.java |  23 +++--
 .../flink/optimizer/dag/BinaryUnionNode.java    |   8 +-
 .../flink/optimizer/UnionReplacementTest.java   | 102 ++++++++++++++++++-
 .../dataexchange/UnionClosedBranchingTest.java  |  26 +++--
 4 files changed, 141 insertions(+), 18 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/a7f6594b/flink-java/src/main/java/org/apache/flink/api/java/operators/OperatorTranslation.java
----------------------------------------------------------------------
diff --git 
a/flink-java/src/main/java/org/apache/flink/api/java/operators/OperatorTranslation.java
 
b/flink-java/src/main/java/org/apache/flink/api/java/operators/OperatorTranslation.java
index 74811a3..2e72137 100644
--- 
a/flink-java/src/main/java/org/apache/flink/api/java/operators/OperatorTranslation.java
+++ 
b/flink-java/src/main/java/org/apache/flink/api/java/operators/OperatorTranslation.java
@@ -41,11 +41,11 @@ import org.apache.flink.configuration.Configuration;
 public class OperatorTranslation {
        
        /** The already translated operations */
-       private Map<DataSet<?>, Operator<?>> translated = new 
HashMap<DataSet<?>, Operator<?>>();
+       private Map<DataSet<?>, Operator<?>> translated = new HashMap<>();
        
        
        public Plan translateToPlan(List<DataSink<?>> sinks, String jobName) {
-               List<GenericDataSinkBase<?>> planSinks = new 
ArrayList<GenericDataSinkBase<?>>();
+               List<GenericDataSinkBase<?>> planSinks = new ArrayList<>();
                
                for (DataSink<?> sink : sinks) {
                        planSinks.add(translate(sink));
@@ -72,11 +72,18 @@ public class OperatorTranslation {
        private <T> Operator<T> translate(DataSet<T> dataSet) {
                
                // check if we have already translated that data set (operation 
or source)
-               Operator<?> previous = (Operator<?>) 
this.translated.get(dataSet);
+               Operator<?> previous = this.translated.get(dataSet);
                if (previous != null) {
-                       @SuppressWarnings("unchecked")
-                       Operator<T> typedPrevious = (Operator<T>) previous;
-                       return typedPrevious;
+
+                       // Union operators may only have a single output.
+                       // We ensure this by not reusing previously created 
union operators.
+                       // The optimizer will merge subsequent binary unions 
into one n-ary union.
+                       if (!(dataSet instanceof UnionOperator)) {
+                               // all other operators are reused.
+                               @SuppressWarnings("unchecked")
+                               Operator<T> typedPrevious = (Operator<T>) 
previous;
+                               return typedPrevious;
+                       }
                }
                
                Operator<T> dataFlowOp;
@@ -188,7 +195,7 @@ public class OperatorTranslation {
                BulkIterationResultSet<T> iterationEnd = 
(BulkIterationResultSet<T>) untypedIterationEnd;
                
                BulkIterationBase<T> iterationOperator =
-                               new BulkIterationBase<T>(new 
UnaryOperatorInformation<T, T>(iterationEnd.getType(), iterationEnd.getType()), 
"Bulk Iteration");
+                               new BulkIterationBase<>(new 
UnaryOperatorInformation<>(iterationEnd.getType(), iterationEnd.getType()), 
"Bulk Iteration");
                IterativeDataSet<T> iterationHead = 
iterationEnd.getIterationHead();
 
                translated.put(iterationHead, 
iterationOperator.getPartialSolution());
@@ -214,7 +221,7 @@ public class OperatorTranslation {
                
                String name = iterationHead.getName() == null ? "Unnamed Delta 
Iteration" : iterationHead.getName();
                
-               DeltaIterationBase<D, W> iterationOperator = new 
DeltaIterationBase<D, W>(new BinaryOperatorInformation<D, W, 
D>(iterationEnd.getType(), iterationEnd.getWorksetType(), 
iterationEnd.getType()),
+               DeltaIterationBase<D, W> iterationOperator = new 
DeltaIterationBase<>(new BinaryOperatorInformation<>(iterationEnd.getType(), 
iterationEnd.getWorksetType(), iterationEnd.getType()),
                                iterationEnd.getKeyPositions(), name);
                
                
iterationOperator.setMaximumNumberOfIterations(iterationEnd.getMaxIterations());

http://git-wip-us.apache.org/repos/asf/flink/blob/a7f6594b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/BinaryUnionNode.java
----------------------------------------------------------------------
diff --git 
a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/BinaryUnionNode.java
 
b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/BinaryUnionNode.java
index fdd76a8..d262cf6 100644
--- 
a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/BinaryUnionNode.java
+++ 
b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/BinaryUnionNode.java
@@ -98,6 +98,12 @@ public class BinaryUnionNode extends TwoInputNode {
        
        @Override
        public List<PlanNode> getAlternativePlans(CostEstimator estimator) {
+
+               // check that union has only a single successor
+               if (this.getOutgoingConnections().size() > 1) {
+                       throw new CompilerException("BinaryUnionNode has more 
than one successor.");
+               }
+
                // check if we have a cached version
                if (this.cachedPlans != null) {
                        return this.cachedPlans;
@@ -173,7 +179,7 @@ public class BinaryUnionNode extends TwoInputNode {
                                                }
                                        }
                                        
-                                       // create a candidate channel for the 
first input. mark it cached, if the connection says so
+                                       // create a candidate channel for the 
second input. mark it cached, if the connection says so
                                        Channel c2 = new Channel(child2, 
this.input2.getMaterializationMode());
                                        if (this.input2.getShipStrategy() == 
null) {
                                                // free to choose the ship 
strategy

http://git-wip-us.apache.org/repos/asf/flink/blob/a7f6594b/flink-optimizer/src/test/java/org/apache/flink/optimizer/UnionReplacementTest.java
----------------------------------------------------------------------
diff --git 
a/flink-optimizer/src/test/java/org/apache/flink/optimizer/UnionReplacementTest.java
 
b/flink-optimizer/src/test/java/org/apache/flink/optimizer/UnionReplacementTest.java
index 65dd2b3..3be7657 100644
--- 
a/flink-optimizer/src/test/java/org/apache/flink/optimizer/UnionReplacementTest.java
+++ 
b/flink-optimizer/src/test/java/org/apache/flink/optimizer/UnionReplacementTest.java
@@ -18,16 +18,25 @@
 
 package org.apache.flink.optimizer;
 
+import junit.framework.Assert;
+import org.apache.flink.api.common.operators.util.FieldList;
 import org.apache.flink.api.java.DataSet;
 import org.apache.flink.api.java.ExecutionEnvironment;
 import org.apache.flink.api.common.Plan;
 import org.apache.flink.api.java.io.DiscardingOutputFormat;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.optimizer.plan.Channel;
+import org.apache.flink.optimizer.plan.NAryUnionPlanNode;
 import org.apache.flink.optimizer.plan.OptimizedPlan;
+import org.apache.flink.optimizer.plan.SingleInputPlanNode;
 import org.apache.flink.optimizer.plantranslate.JobGraphGenerator;
 import org.apache.flink.optimizer.util.CompilerTestBase;
+import org.apache.flink.runtime.operators.shipping.ShipStrategyType;
 import org.junit.Test;
 
-import static org.junit.Assert.fail;
+import java.util.List;
+
+import static org.junit.Assert.*;
 
 @SuppressWarnings("serial")
 public class UnionReplacementTest extends CompilerTestBase {
@@ -54,4 +63,95 @@ public class UnionReplacementTest extends CompilerTestBase {
                        fail(e.getMessage());
                }
        }
+
+       /**
+        *
+        * Test for FLINK-2662.
+        *
+        * Checks that a plan with an union with two outputs is correctly 
translated.
+        * The program can be illustrated as follows:
+        *
+        * Src1 ----------------\
+        *                       >-> Union123 -> GroupBy(0) -> Sum -> Output
+        * Src2 -\              /
+        *        >-> Union23--<
+        * Src3 -/              \
+        *                       >-> Union234 -> GroupBy(1) -> Sum -> Output
+        * Src4 ----------------/
+        *
+        * The fix for FLINK-2662 translates the union with two output 
(Union-23) into two separate
+        * unions (Union-23_1 and Union-23_2) with one output each. Due to this 
change, the interesting
+        * partitioning properties for GroupBy(0) and GroupBy(1) are pushed 
through Union-23_1 and
+        * Union-23_2 and do not interfere with each other (which would be the 
case if Union-23 would
+        * be a single operator with two outputs).
+        *
+        */
+       @Test
+       public void testUnionWithTwoOutputsTest() throws Exception {
+
+               // 
-----------------------------------------------------------------------------------------
+               // Build test program
+               // 
-----------------------------------------------------------------------------------------
+
+               ExecutionEnvironment env = 
ExecutionEnvironment.getExecutionEnvironment();
+               env.setParallelism(DEFAULT_PARALLELISM);
+
+               DataSet<Tuple2<Long, Long>> src1 = env.fromElements(new 
Tuple2<>(0L, 0L));
+               DataSet<Tuple2<Long, Long>> src2 = env.fromElements(new 
Tuple2<>(0L, 0L));
+               DataSet<Tuple2<Long, Long>> src3 = env.fromElements(new 
Tuple2<>(0L, 0L));
+               DataSet<Tuple2<Long, Long>> src4 = env.fromElements(new 
Tuple2<>(0L, 0L));
+
+               DataSet<Tuple2<Long, Long>> union23 = src2.union(src3);
+               DataSet<Tuple2<Long, Long>> union123 = src1.union(union23);
+               DataSet<Tuple2<Long, Long>> union234 = src4.union(union23);
+
+               union123.groupBy(0).sum(1).name("1").output(new 
DiscardingOutputFormat<Tuple2<Long, Long>>());
+               union234.groupBy(1).sum(0).name("2").output(new 
DiscardingOutputFormat<Tuple2<Long, Long>>());
+
+               // 
-----------------------------------------------------------------------------------------
+               // Verify optimized plan
+               // 
-----------------------------------------------------------------------------------------
+
+               OptimizedPlan optimizedPlan = 
compileNoStats(env.createProgramPlan());
+
+               OptimizerPlanNodeResolver resolver = 
getOptimizerPlanNodeResolver(optimizedPlan);
+
+               SingleInputPlanNode groupRed1 = resolver.getNode("1");
+               SingleInputPlanNode groupRed2 = resolver.getNode("2");
+
+               // check partitioning is correct
+               Assert.assertTrue("Reduce input should be partitioned on 0.",
+                       
groupRed1.getInput().getGlobalProperties().getPartitioningFields().isExactMatch(new
 FieldList(0)));
+               Assert.assertTrue("Reduce input should be partitioned on 1.",
+                       
groupRed2.getInput().getGlobalProperties().getPartitioningFields().isExactMatch(new
 FieldList(1)));
+
+               // check group reduce inputs are n-ary unions with three inputs
+               Assert.assertTrue("Reduce input should be n-ary union with 
three inputs.",
+                       groupRed1.getInput().getSource() instanceof 
NAryUnionPlanNode &&
+                               ((NAryUnionPlanNode) 
groupRed1.getInput().getSource()).getListOfInputs().size() == 3);
+               Assert.assertTrue("Reduce input should be n-ary union with 
three inputs.",
+                       groupRed2.getInput().getSource() instanceof 
NAryUnionPlanNode &&
+                               ((NAryUnionPlanNode) 
groupRed2.getInput().getSource()).getListOfInputs().size() == 3);
+
+               // check channel from union to group reduce is forwarding
+               Assert.assertTrue("Channel between union and group reduce 
should be forwarding",
+                       
groupRed1.getInput().getShipStrategy().equals(ShipStrategyType.FORWARD));
+               Assert.assertTrue("Channel between union and group reduce 
should be forwarding",
+                       
groupRed2.getInput().getShipStrategy().equals(ShipStrategyType.FORWARD));
+
+               // check that all inputs of unions are hash partitioned
+               List<Channel> union123In = ((NAryUnionPlanNode) 
groupRed1.getInput().getSource()).getListOfInputs();
+               for(Channel i : union123In) {
+                       Assert.assertTrue("Union input channel should hash 
partition on 0",
+                               
i.getShipStrategy().equals(ShipStrategyType.PARTITION_HASH) &&
+                                       
i.getShipStrategyKeys().isExactMatch(new FieldList(0)));
+               }
+               List<Channel> union234In = ((NAryUnionPlanNode) 
groupRed2.getInput().getSource()).getListOfInputs();
+               for(Channel i : union234In) {
+                       Assert.assertTrue("Union input channel should hash 
partition on 0",
+                               
i.getShipStrategy().equals(ShipStrategyType.PARTITION_HASH) &&
+                                       
i.getShipStrategyKeys().isExactMatch(new FieldList(1)));
+               }
+
+       }
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/a7f6594b/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataexchange/UnionClosedBranchingTest.java
----------------------------------------------------------------------
diff --git 
a/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataexchange/UnionClosedBranchingTest.java
 
b/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataexchange/UnionClosedBranchingTest.java
index b870a91..77b150a 100644
--- 
a/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataexchange/UnionClosedBranchingTest.java
+++ 
b/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataexchange/UnionClosedBranchingTest.java
@@ -37,6 +37,7 @@ import 
org.apache.flink.runtime.io.network.partition.consumer.UnionInputGate;
 import org.apache.flink.runtime.jobgraph.IntermediateDataSet;
 import org.apache.flink.runtime.jobgraph.JobGraph;
 import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.runtime.operators.shipping.ShipStrategyType;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
@@ -73,9 +74,9 @@ public class UnionClosedBranchingTest extends 
CompilerTestBase {
        @Parameterized.Parameters
        public static Collection<Object[]> params() {
                Collection<Object[]> params = Arrays.asList(new Object[][]{
-                               {ExecutionMode.PIPELINED, PIPELINED, BATCH},
+                               {ExecutionMode.PIPELINED, BATCH, PIPELINED},
                                {ExecutionMode.PIPELINED_FORCED, PIPELINED, 
PIPELINED},
-                               {ExecutionMode.BATCH, BATCH, BATCH},
+                               {ExecutionMode.BATCH, BATCH, PIPELINED},
                                {ExecutionMode.BATCH_FORCED, BATCH, BATCH},
                });
 
@@ -93,10 +94,16 @@ public class UnionClosedBranchingTest extends 
CompilerTestBase {
        /** Expected {@link DataExchangeMode} from union to join. */
        private final DataExchangeMode unionToJoin;
 
+       /** Expected {@link ShipStrategyType} from source to union. */
+       private final ShipStrategyType sourceToUnionStrategy = 
ShipStrategyType.PARTITION_HASH;
+
+       /** Expected {@link ShipStrategyType} from union to join. */
+       private final ShipStrategyType unionToJoinStrategy = 
ShipStrategyType.FORWARD;
+
        public UnionClosedBranchingTest(
-                       ExecutionMode executionMode,
-                       DataExchangeMode sourceToUnion,
-                       DataExchangeMode unionToJoin) {
+               ExecutionMode executionMode,
+               DataExchangeMode sourceToUnion,
+               DataExchangeMode unionToJoin) {
 
                this.executionMode = executionMode;
                this.sourceToUnion = sourceToUnion;
@@ -140,12 +147,16 @@ public class UnionClosedBranchingTest extends 
CompilerTestBase {
                for (Channel channel : joinNode.getInputs()) {
                        assertEquals("Unexpected data exchange mode between 
union and join node.",
                                        unionToJoin, 
channel.getDataExchangeMode());
+                       assertEquals("Unexpected ship strategy between union 
and join node.",
+                                       unionToJoinStrategy, 
channel.getShipStrategy());
                }
 
                for (SourcePlanNode src : optimizedPlan.getDataSources()) {
                        for (Channel channel : src.getOutgoingChannels()) {
                                assertEquals("Unexpected data exchange mode 
between source and union node.",
                                                sourceToUnion, 
channel.getDataExchangeMode());
+                               assertEquals("Unexpected ship strategy between 
source and union node.",
+                                       sourceToUnionStrategy, 
channel.getShipStrategy());
                        }
                }
 
@@ -176,9 +187,8 @@ public class UnionClosedBranchingTest extends 
CompilerTestBase {
                        for (IntermediateDataSet dataSet : 
src.getProducedDataSets()) {
                                ResultPartitionType dsType = 
dataSet.getResultType();
 
-                               // The result type is determined by the channel 
between the union and the join node
-                               // and *not* the channel between source and 
union.
-                               if (unionToJoin.equals(BATCH)) {
+                               // Ensure batch exchange unless PIPELINED_FORCE 
is enabled.
+                               if 
(!executionMode.equals(ExecutionMode.PIPELINED_FORCED)) {
                                        assertTrue("Expected batch exchange, 
but result type is " + dsType + ".",
                                                        dsType.isBlocking());
                                } else {

Reply via email to