You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by rm...@apache.org on 2014/10/18 19:47:00 UTC
[1/2] git commit: [FLINK-1167] Handle unions at the root of the
iteration
Repository: incubator-flink
Updated Branches:
refs/heads/release-0.7 ce2936ab1 -> 58943adb9
[FLINK-1167] Handle unions at the root of the iteration
This closes #160
Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/58943adb
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/58943adb
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/58943adb
Branch: refs/heads/release-0.7
Commit: 58943adb9bfd2f95b4c35c5ecf92059a0d0417c6
Parents: 50f5ea8
Author: Stephan Ewen <se...@apache.org>
Authored: Sat Oct 18 03:08:39 2014 +0200
Committer: Robert Metzger <rm...@apache.org>
Committed: Sat Oct 18 19:46:29 2014 +0200
----------------------------------------------------------------------
.../flink/compiler/dag/BulkIterationNode.java | 2 +-
.../compiler/dag/WorksetIterationNode.java | 2 +-
.../plandump/PlanJSONDumpGenerator.java | 186 +++++++++----------
.../compiler/java/IterationCompilerTest.java | 62 ++++++-
4 files changed, 144 insertions(+), 108 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/58943adb/flink-compiler/src/main/java/org/apache/flink/compiler/dag/BulkIterationNode.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/main/java/org/apache/flink/compiler/dag/BulkIterationNode.java b/flink-compiler/src/main/java/org/apache/flink/compiler/dag/BulkIterationNode.java
index a5f8026..d3f0fbb 100644
--- a/flink-compiler/src/main/java/org/apache/flink/compiler/dag/BulkIterationNode.java
+++ b/flink-compiler/src/main/java/org/apache/flink/compiler/dag/BulkIterationNode.java
@@ -132,7 +132,7 @@ public class BulkIterationNode extends SingleInputNode implements IterationNode
// check if the root of the step function has the same DOP as the iteration
// or if the steo function has any operator at all
if (nextPartialSolution.getDegreeOfParallelism() != getDegreeOfParallelism() ||
- nextPartialSolution == partialSolution)
+ nextPartialSolution == partialSolution || nextPartialSolution instanceof BinaryUnionNode)
{
// add a no-op to the root to express the re-partitioning
NoOpNode noop = new NoOpNode();
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/58943adb/flink-compiler/src/main/java/org/apache/flink/compiler/dag/WorksetIterationNode.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/main/java/org/apache/flink/compiler/dag/WorksetIterationNode.java b/flink-compiler/src/main/java/org/apache/flink/compiler/dag/WorksetIterationNode.java
index 7638cca..b6ae34e 100644
--- a/flink-compiler/src/main/java/org/apache/flink/compiler/dag/WorksetIterationNode.java
+++ b/flink-compiler/src/main/java/org/apache/flink/compiler/dag/WorksetIterationNode.java
@@ -160,7 +160,7 @@ public class WorksetIterationNode extends TwoInputNode implements IterationNode
// there needs to be at least one node in the workset path, so
// if the next workset is equal to the workset, we need to inject a no-op node
- if (nextWorkset == worksetNode) {
+ if (nextWorkset == worksetNode || nextWorkset instanceof BinaryUnionNode) {
NoOpNode noop = new NoOpNode();
noop.setDegreeOfParallelism(getDegreeOfParallelism());
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/58943adb/flink-compiler/src/main/java/org/apache/flink/compiler/plandump/PlanJSONDumpGenerator.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/main/java/org/apache/flink/compiler/plandump/PlanJSONDumpGenerator.java b/flink-compiler/src/main/java/org/apache/flink/compiler/plandump/PlanJSONDumpGenerator.java
index 00e2bc2..60500b8 100644
--- a/flink-compiler/src/main/java/org/apache/flink/compiler/plandump/PlanJSONDumpGenerator.java
+++ b/flink-compiler/src/main/java/org/apache/flink/compiler/plandump/PlanJSONDumpGenerator.java
@@ -16,7 +16,6 @@
* limitations under the License.
*/
-
package org.apache.flink.compiler.plandump;
import java.io.File;
@@ -26,7 +25,6 @@ import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.ArrayList;
import java.util.Collection;
-import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
@@ -47,7 +45,6 @@ import org.apache.flink.compiler.dataproperties.GlobalProperties;
import org.apache.flink.compiler.dataproperties.LocalProperties;
import org.apache.flink.compiler.plan.BulkIterationPlanNode;
import org.apache.flink.compiler.plan.Channel;
-import org.apache.flink.compiler.plan.NAryUnionPlanNode;
import org.apache.flink.compiler.plan.OptimizedPlan;
import org.apache.flink.compiler.plan.PlanNode;
import org.apache.flink.compiler.plan.SingleInputPlanNode;
@@ -265,121 +262,104 @@ public class PlanJSONDumpGenerator {
if (inConns != null && inConns.hasNext()) {
// start predecessor list
writer.print(",\n\t\t\"predecessors\": [");
- int connNum = 0;
int inputNum = 0;
while (inConns.hasNext()) {
- final DumpableConnection<?> conn = inConns.next();
-
- final Collection<DumpableConnection<?>> inConnsForInput;
- if (conn.getSource() instanceof NAryUnionPlanNode) {
- inConnsForInput = new ArrayList<DumpableConnection<?>>();
+ final DumpableConnection<?> inConn = inConns.next();
+ final DumpableNode<?> source = inConn.getSource();
+ writer.print(inputNum == 0 ? "\n" : ",\n");
+ if (inputNum == 0) {
+ child1name += child1name.length() > 0 ? ", " : "";
+ child1name += source.getOptimizerNode().getPactContract().getName();
+ } else if (inputNum == 1) {
+ child2name += child2name.length() > 0 ? ", " : "";
+ child2name = source.getOptimizerNode().getPactContract().getName();
+ }
+
+ // output predecessor id
+ writer.print("\t\t\t{\"id\": " + this.nodeIds.get(source));
+
+ // output connection side
+ if (inConns.hasNext() || inputNum > 0) {
+ writer.print(", \"side\": \"" + (inputNum == 0 ? "first" : "second") + "\"");
+ }
+ // output shipping strategy and channel type
+ final Channel channel = (inConn instanceof Channel) ? (Channel) inConn : null;
+ final ShipStrategyType shipType = channel != null ? channel.getShipStrategy() :
+ ((PactConnection) inConn).getShipStrategy();
- for (DumpableConnection<?> inputOfUnion : conn.getSource().getDumpableInputs()) {
- inConnsForInput.add(inputOfUnion);
+ String shipStrategy = null;
+ if (shipType != null) {
+ switch (shipType) {
+ case NONE:
+ // nothing
+ break;
+ case FORWARD:
+ shipStrategy = "Forward";
+ break;
+ case BROADCAST:
+ shipStrategy = "Broadcast";
+ break;
+ case PARTITION_HASH:
+ shipStrategy = "Hash Partition";
+ break;
+ case PARTITION_RANGE:
+ shipStrategy = "Range Partition";
+ break;
+ case PARTITION_RANDOM:
+ shipStrategy = "Redistribute";
+ break;
+ case PARTITION_FORCED_REBALANCE:
+ shipStrategy = "Rebalance";
+ break;
+ default:
+ throw new CompilerException("Unknown ship strategy '" + inConn.getShipStrategy().name()
+ + "' in JSON generator.");
}
}
- else {
- inConnsForInput = Collections.<DumpableConnection<?>>singleton(conn);
+
+ if (channel != null && channel.getShipStrategyKeys() != null && channel.getShipStrategyKeys().size() > 0) {
+ shipStrategy += " on " + (channel.getShipStrategySortOrder() == null ?
+ channel.getShipStrategyKeys().toString() :
+ Utils.createOrdering(channel.getShipStrategyKeys(), channel.getShipStrategySortOrder()).toString());
+ }
+
+ if (shipStrategy != null) {
+ writer.print(", \"ship_strategy\": \"" + shipStrategy + "\"");
}
- for (DumpableConnection<?> inConn : inConnsForInput) {
- final DumpableNode<?> source = inConn.getSource();
- writer.print(connNum == 0 ? "\n" : ",\n");
- if (connNum == 0) {
- child1name += child1name.length() > 0 ? ", " : "";
- child1name += source.getOptimizerNode().getPactContract().getName();
- } else if (connNum == 1) {
- child2name += child2name.length() > 0 ? ", " : "";
- child2name = source.getOptimizerNode().getPactContract().getName();
- }
-
- // output predecessor id
- writer.print("\t\t\t{\"id\": " + this.nodeIds.get(source));
-
- // output connection side
- if (inConns.hasNext() || inputNum > 0) {
- writer.print(", \"side\": \"" + (inputNum == 0 ? "first" : "second") + "\"");
- }
- // output shipping strategy and channel type
- final Channel channel = (inConn instanceof Channel) ? (Channel) inConn : null;
- final ShipStrategyType shipType = channel != null ? channel.getShipStrategy() :
- ((PactConnection) inConn).getShipStrategy();
-
- String shipStrategy = null;
- if (shipType != null) {
- switch (shipType) {
- case NONE:
- // nothing
- break;
- case FORWARD:
- shipStrategy = "Forward";
- break;
- case BROADCAST:
- shipStrategy = "Broadcast";
- break;
- case PARTITION_HASH:
- shipStrategy = "Hash Partition";
- break;
- case PARTITION_RANGE:
- shipStrategy = "Range Partition";
- break;
- case PARTITION_RANDOM:
- shipStrategy = "Redistribute";
- break;
- case PARTITION_FORCED_REBALANCE:
- shipStrategy = "Rebalance";
- break;
- default:
- throw new CompilerException("Unknown ship strategy '" + conn.getShipStrategy().name()
- + "' in JSON generator.");
- }
+ if (channel != null) {
+ String localStrategy = null;
+ switch (channel.getLocalStrategy()) {
+ case NONE:
+ break;
+ case SORT:
+ localStrategy = "Sort";
+ break;
+ case COMBININGSORT:
+ localStrategy = "Sort (combining)";
+ break;
+ default:
+ throw new CompilerException("Unknown local strategy " + channel.getLocalStrategy().name());
}
- if (channel != null && channel.getShipStrategyKeys() != null && channel.getShipStrategyKeys().size() > 0) {
- shipStrategy += " on " + (channel.getShipStrategySortOrder() == null ?
- channel.getShipStrategyKeys().toString() :
- Utils.createOrdering(channel.getShipStrategyKeys(), channel.getShipStrategySortOrder()).toString());
- }
-
- if (shipStrategy != null) {
- writer.print(", \"ship_strategy\": \"" + shipStrategy + "\"");
+ if (channel != null && channel.getLocalStrategyKeys() != null && channel.getLocalStrategyKeys().size() > 0) {
+ localStrategy += " on " + (channel.getLocalStrategySortOrder() == null ?
+ channel.getLocalStrategyKeys().toString() :
+ Utils.createOrdering(channel.getLocalStrategyKeys(), channel.getLocalStrategySortOrder()).toString());
}
- if (channel != null) {
- String localStrategy = null;
- switch (channel.getLocalStrategy()) {
- case NONE:
- break;
- case SORT:
- localStrategy = "Sort";
- break;
- case COMBININGSORT:
- localStrategy = "Sort (combining)";
- break;
- default:
- throw new CompilerException("Unknown local strategy " + channel.getLocalStrategy().name());
- }
-
- if (channel != null && channel.getLocalStrategyKeys() != null && channel.getLocalStrategyKeys().size() > 0) {
- localStrategy += " on " + (channel.getLocalStrategySortOrder() == null ?
- channel.getLocalStrategyKeys().toString() :
- Utils.createOrdering(channel.getLocalStrategyKeys(), channel.getLocalStrategySortOrder()).toString());
- }
-
- if (localStrategy != null) {
- writer.print(", \"local_strategy\": \"" + localStrategy + "\"");
- }
-
- if (channel != null && channel.getTempMode() != TempMode.NONE) {
- String tempMode = channel.getTempMode().toString();
- writer.print(", \"temp_mode\": \"" + tempMode + "\"");
- }
+ if (localStrategy != null) {
+ writer.print(", \"local_strategy\": \"" + localStrategy + "\"");
}
- writer.print('}');
- connNum++;
+ if (channel != null && channel.getTempMode() != TempMode.NONE) {
+ String tempMode = channel.getTempMode().toString();
+ writer.print(", \"temp_mode\": \"" + tempMode + "\"");
+ }
}
+
+ writer.print('}');
inputNum++;
}
// finish predecessors
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/58943adb/flink-compiler/src/test/java/org/apache/flink/compiler/java/IterationCompilerTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/compiler/java/IterationCompilerTest.java b/flink-compiler/src/test/java/org/apache/flink/compiler/java/IterationCompilerTest.java
index c3fbbf2..36e5739 100644
--- a/flink-compiler/src/test/java/org/apache/flink/compiler/java/IterationCompilerTest.java
+++ b/flink-compiler/src/test/java/org/apache/flink/compiler/java/IterationCompilerTest.java
@@ -29,8 +29,8 @@ import org.apache.flink.api.java.operators.IterativeDataSet;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.compiler.CompilerTestBase;
import org.apache.flink.compiler.plan.OptimizedPlan;
-import org.apache.flink.compiler.plandump.PlanJSONDumpGenerator;
import org.apache.flink.compiler.plantranslate.NepheleJobGraphGenerator;
+import org.apache.flink.compiler.testfunctions.IdentityMapper;
import org.junit.Test;
@SuppressWarnings("serial")
@@ -57,7 +57,7 @@ public class IterationCompilerTest extends CompilerTestBase {
}
@Test
- public void testIdentityWorksetIteration() {
+ public void testEmptyWorksetIteration() {
try {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
env.setDegreeOfParallelism(43);
@@ -76,7 +76,63 @@ public class IterationCompilerTest extends CompilerTestBase {
Plan p = env.createProgramPlan();
OptimizedPlan op = compileNoStats(p);
- System.out.println(new PlanJSONDumpGenerator().getOptimizerPlanAsJSON(op));
+ new NepheleJobGraphGenerator().compileJobGraph(op);
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testIterationWithUnionRoot() {
+ try {
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ env.setDegreeOfParallelism(43);
+
+ IterativeDataSet<Long> iteration = env.generateSequence(-4, 1000).iterate(100);
+
+ iteration.closeWith(
+ iteration.map(new IdentityMapper<Long>()).union(iteration.map(new IdentityMapper<Long>())))
+ .print();
+
+ Plan p = env.createProgramPlan();
+ OptimizedPlan op = compileNoStats(p);
+
+ new NepheleJobGraphGenerator().compileJobGraph(op);
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testWorksetIterationWithUnionRoot() {
+ try {
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ env.setDegreeOfParallelism(43);
+
+ DataSet<Tuple2<Long, Long>> input = env.generateSequence(1, 20)
+ .map(new MapFunction<Long, Tuple2<Long, Long>>() {
+ @Override
+ public Tuple2<Long, Long> map(Long value){ return null; }
+ });
+
+
+ DeltaIteration<Tuple2<Long, Long>, Tuple2<Long, Long>> iter = input.iterateDelta(input, 100, 0);
+ iter.closeWith(
+ iter.getWorkset().map(new IdentityMapper<Tuple2<Long,Long>>())
+ .union(
+ iter.getWorkset().map(new IdentityMapper<Tuple2<Long,Long>>()))
+ , iter.getWorkset().map(new IdentityMapper<Tuple2<Long,Long>>())
+ .union(
+ iter.getWorkset().map(new IdentityMapper<Tuple2<Long,Long>>()))
+ )
+ .print();
+
+ Plan p = env.createProgramPlan();
+ OptimizedPlan op = compileNoStats(p);
new NepheleJobGraphGenerator().compileJobGraph(op);
}
[2/2] git commit: [FLINK-1164] Gracefully handle empty (identity)
iterations
Posted by rm...@apache.org.
[FLINK-1164] Gracefully handle empty (identity) iterations
Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/50f5ea83
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/50f5ea83
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/50f5ea83
Branch: refs/heads/release-0.7
Commit: 50f5ea83c20d1b191c64f28ca6653dd79b05b8f5
Parents: ce2936a
Author: Stephan Ewen <se...@apache.org>
Authored: Sat Oct 18 01:55:08 2014 +0200
Committer: Robert Metzger <rm...@apache.org>
Committed: Sat Oct 18 19:46:29 2014 +0200
----------------------------------------------------------------------
.../org/apache/flink/compiler/PactCompiler.java | 5 +-
.../flink/compiler/dag/BulkIterationNode.java | 4 +-
.../compiler/dag/WorksetIterationNode.java | 15 +++-
.../plantranslate/NepheleJobGraphGenerator.java | 10 +++
.../compiler/java/IterationCompilerTest.java | 88 ++++++++++++++++++++
.../iterative/EmptyWorksetIterationITCase.java | 59 +++++++++++++
.../test/iterative/IdentityIterationITCase.java | 56 +++++++++++++
7 files changed, 233 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/50f5ea83/flink-compiler/src/main/java/org/apache/flink/compiler/PactCompiler.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/main/java/org/apache/flink/compiler/PactCompiler.java b/flink-compiler/src/main/java/org/apache/flink/compiler/PactCompiler.java
index 177622b..d8792c0 100644
--- a/flink-compiler/src/main/java/org/apache/flink/compiler/PactCompiler.java
+++ b/flink-compiler/src/main/java/org/apache/flink/compiler/PactCompiler.java
@@ -801,7 +801,8 @@ public class PactCompiler {
final BulkIterationNode iterNode = (BulkIterationNode) n;
final BulkIterationBase<?> iter = iterNode.getIterationContract();
- // calculate closure of the anonymous function
+ // pass a copy of the no iterative part into the iteration translation,
+ // in case the iteration references its closure
HashMap<Operator<?>, OptimizerNode> closure = new HashMap<Operator<?>, OptimizerNode>(con2node);
// first, recursively build the data flow for the step function
@@ -831,8 +832,8 @@ public class PactCompiler {
}
}
- iterNode.setNextPartialSolution(rootOfStepFunction, terminationCriterion);
iterNode.setPartialSolution(partialSolution);
+ iterNode.setNextPartialSolution(rootOfStepFunction, terminationCriterion);
// go over the contained data flow and mark the dynamic path nodes
StaticDynamicPathIdentifier identifier = new StaticDynamicPathIdentifier(iterNode.getCostWeight());
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/50f5ea83/flink-compiler/src/main/java/org/apache/flink/compiler/dag/BulkIterationNode.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/main/java/org/apache/flink/compiler/dag/BulkIterationNode.java b/flink-compiler/src/main/java/org/apache/flink/compiler/dag/BulkIterationNode.java
index 91e9eda..a5f8026 100644
--- a/flink-compiler/src/main/java/org/apache/flink/compiler/dag/BulkIterationNode.java
+++ b/flink-compiler/src/main/java/org/apache/flink/compiler/dag/BulkIterationNode.java
@@ -130,7 +130,9 @@ public class BulkIterationNode extends SingleInputNode implements IterationNode
public void setNextPartialSolution(OptimizerNode nextPartialSolution, OptimizerNode terminationCriterion) {
// check if the root of the step function has the same DOP as the iteration
- if (nextPartialSolution.getDegreeOfParallelism() != getDegreeOfParallelism())
+ // or if the steo function has any operator at all
+ if (nextPartialSolution.getDegreeOfParallelism() != getDegreeOfParallelism() ||
+ nextPartialSolution == partialSolution)
{
// add a no-op to the root to express the re-partitioning
NoOpNode noop = new NoOpNode();
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/50f5ea83/flink-compiler/src/main/java/org/apache/flink/compiler/dag/WorksetIterationNode.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/main/java/org/apache/flink/compiler/dag/WorksetIterationNode.java b/flink-compiler/src/main/java/org/apache/flink/compiler/dag/WorksetIterationNode.java
index 2cd8294..7638cca 100644
--- a/flink-compiler/src/main/java/org/apache/flink/compiler/dag/WorksetIterationNode.java
+++ b/flink-compiler/src/main/java/org/apache/flink/compiler/dag/WorksetIterationNode.java
@@ -158,6 +158,19 @@ public class WorksetIterationNode extends TwoInputNode implements IterationNode
}
}
+ // there needs to be at least one node in the workset path, so
+ // if the next workset is equal to the workset, we need to inject a no-op node
+ if (nextWorkset == worksetNode) {
+ NoOpNode noop = new NoOpNode();
+ noop.setDegreeOfParallelism(getDegreeOfParallelism());
+
+ PactConnection noOpConn = new PactConnection(nextWorkset, noop);
+ noop.setIncomingConnection(noOpConn);
+ nextWorkset.addOutgoingConnection(noOpConn);
+
+ nextWorkset = noop;
+ }
+
// attach an extra node to the solution set delta for the cases where we need to repartition
UnaryOperatorNode solutionSetDeltaUpdateAux = new UnaryOperatorNode("Solution-Set Delta", getSolutionSetKeyFields(),
new SolutionSetDeltaOperator(getSolutionSetKeyFields()));
@@ -367,7 +380,7 @@ public class WorksetIterationNode extends TwoInputNode implements IterationNode
return;
}
- // sanity check the solution set delta and cancel out the delta node, if it is not needed
+ // sanity check the solution set delta
for (Iterator<PlanNode> deltaPlans = solutionSetDeltaCandidates.iterator(); deltaPlans.hasNext(); ) {
SingleInputPlanNode candidate = (SingleInputPlanNode) deltaPlans.next();
GlobalProperties gp = candidate.getGlobalProperties();
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/50f5ea83/flink-compiler/src/main/java/org/apache/flink/compiler/plantranslate/NepheleJobGraphGenerator.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/main/java/org/apache/flink/compiler/plantranslate/NepheleJobGraphGenerator.java b/flink-compiler/src/main/java/org/apache/flink/compiler/plantranslate/NepheleJobGraphGenerator.java
index dbce56b..3dd9685 100644
--- a/flink-compiler/src/main/java/org/apache/flink/compiler/plantranslate/NepheleJobGraphGenerator.java
+++ b/flink-compiler/src/main/java/org/apache/flink/compiler/plantranslate/NepheleJobGraphGenerator.java
@@ -1129,6 +1129,11 @@ public class NepheleJobGraphGenerator implements Visitor<PlanNode> {
// ------------ finalize the head config with the final outputs and the sync gate ------------
final int numStepFunctionOuts = headConfig.getNumOutputs();
final int numFinalOuts = headFinalOutputConfig.getNumOutputs();
+
+ if (numStepFunctionOuts == 0) {
+ throw new CompilerException("The iteration has no operation inside the step function.");
+ }
+
headConfig.setIterationHeadFinalOutputConfig(headFinalOutputConfig);
headConfig.setIterationHeadIndexOfSyncOutput(numStepFunctionOuts + numFinalOuts);
final double relativeMemForBackChannel = bulkNode.getRelativeMemoryPerSubTask();
@@ -1250,6 +1255,11 @@ public class NepheleJobGraphGenerator implements Visitor<PlanNode> {
{
final int numStepFunctionOuts = headConfig.getNumOutputs();
final int numFinalOuts = headFinalOutputConfig.getNumOutputs();
+
+ if (numStepFunctionOuts == 0) {
+ throw new CompilerException("The workset iteration has no operation on the workset inside the step function.");
+ }
+
headConfig.setIterationHeadFinalOutputConfig(headFinalOutputConfig);
headConfig.setIterationHeadIndexOfSyncOutput(numStepFunctionOuts + numFinalOuts);
final double relativeMemory = iterNode.getRelativeMemoryPerSubTask();
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/50f5ea83/flink-compiler/src/test/java/org/apache/flink/compiler/java/IterationCompilerTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/compiler/java/IterationCompilerTest.java b/flink-compiler/src/test/java/org/apache/flink/compiler/java/IterationCompilerTest.java
new file mode 100644
index 0000000..c3fbbf2
--- /dev/null
+++ b/flink-compiler/src/test/java/org/apache/flink/compiler/java/IterationCompilerTest.java
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.compiler.java;
+
+import static org.junit.Assert.*;
+
+import org.apache.flink.api.common.Plan;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.operators.DeltaIteration;
+import org.apache.flink.api.java.operators.IterativeDataSet;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.compiler.CompilerTestBase;
+import org.apache.flink.compiler.plan.OptimizedPlan;
+import org.apache.flink.compiler.plandump.PlanJSONDumpGenerator;
+import org.apache.flink.compiler.plantranslate.NepheleJobGraphGenerator;
+import org.junit.Test;
+
+@SuppressWarnings("serial")
+public class IterationCompilerTest extends CompilerTestBase {
+
+ @Test
+ public void testIdentityIteration() {
+ try {
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ env.setDegreeOfParallelism(43);
+
+ IterativeDataSet<Long> iteration = env.generateSequence(-4, 1000).iterate(100);
+ iteration.closeWith(iteration).print();
+
+ Plan p = env.createProgramPlan();
+ OptimizedPlan op = compileNoStats(p);
+
+ new NepheleJobGraphGenerator().compileJobGraph(op);
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testIdentityWorksetIteration() {
+ try {
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ env.setDegreeOfParallelism(43);
+
+ DataSet<Tuple2<Long, Long>> input = env.generateSequence(1, 20)
+ .map(new MapFunction<Long, Tuple2<Long, Long>>() {
+ @Override
+ public Tuple2<Long, Long> map(Long value){ return null; }
+ });
+
+
+ DeltaIteration<Tuple2<Long, Long>, Tuple2<Long, Long>> iter = input.iterateDelta(input, 100, 0);
+ iter.closeWith(iter.getWorkset(), iter.getWorkset())
+ .print();
+
+ Plan p = env.createProgramPlan();
+ OptimizedPlan op = compileNoStats(p);
+
+ System.out.println(new PlanJSONDumpGenerator().getOptimizerPlanAsJSON(op));
+
+ new NepheleJobGraphGenerator().compileJobGraph(op);
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/50f5ea83/flink-tests/src/test/java/org/apache/flink/test/iterative/EmptyWorksetIterationITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/iterative/EmptyWorksetIterationITCase.java b/flink-tests/src/test/java/org/apache/flink/test/iterative/EmptyWorksetIterationITCase.java
new file mode 100644
index 0000000..3f02064
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/iterative/EmptyWorksetIterationITCase.java
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.test.iterative;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.io.LocalCollectionOutputFormat;
+import org.apache.flink.api.java.operators.DeltaIteration;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.test.util.JavaProgramTestBase;
+
+@SuppressWarnings("serial")
+public class EmptyWorksetIterationITCase extends JavaProgramTestBase {
+
+ private List<Tuple2<Long, Long>> result = new ArrayList<Tuple2<Long, Long>>();
+
+ @Override
+ protected void testProgram() throws Exception {
+
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ DataSet<Tuple2<Long, Long>> input = env.generateSequence(1, 20).map(new Dupl());
+
+ DeltaIteration<Tuple2<Long, Long>, Tuple2<Long, Long>> iter = input.iterateDelta(input, 20, 0);
+ iter.closeWith(iter.getWorkset(), iter.getWorkset())
+ .output(new LocalCollectionOutputFormat<Tuple2<Long, Long>>(result));
+
+ env.execute();
+ }
+
+ public static final class Dupl implements MapFunction<Long, Tuple2<Long, Long>> {
+
+ @Override
+ public Tuple2<Long, Long> map(Long value) {
+ return new Tuple2<Long, Long>(value, value);
+ }
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/50f5ea83/flink-tests/src/test/java/org/apache/flink/test/iterative/IdentityIterationITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/iterative/IdentityIterationITCase.java b/flink-tests/src/test/java/org/apache/flink/test/iterative/IdentityIterationITCase.java
new file mode 100644
index 0000000..c422aa4
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/iterative/IdentityIterationITCase.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.test.iterative;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.io.LocalCollectionOutputFormat;
+import org.apache.flink.api.java.operators.IterativeDataSet;
+import org.apache.flink.test.util.JavaProgramTestBase;
+
+public class IdentityIterationITCase extends JavaProgramTestBase {
+
+ private List<Long> result = new ArrayList<Long>();
+
+ @Override
+ protected void testProgram() throws Exception {
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ IterativeDataSet<Long> iteration = env.generateSequence(1, 10).iterate(100);
+ iteration.closeWith(iteration)
+ .output(new LocalCollectionOutputFormat<Long>(result));
+
+ env.execute();
+ }
+
+ @Override
+ protected void postSubmit() {
+ assertEquals(10, result.size());
+
+ long sum = 0;
+ for (Long l : result) {
+ sum += l;
+ }
+ assertEquals(55, sum);
+ }
+}