You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tez.apache.org by ss...@apache.org on 2013/10/24 07:04:12 UTC
git commit: TEZ-577. Ensure Input names on a Vertex do not collide
with InputVertex Names for the vertex. (sseth)
Updated Branches:
refs/heads/master 66746073c -> 3beb9f72b
TEZ-577. Ensure Input names on a Vertex do not collide with InputVertex
Names for the vertex. (sseth)
Project: http://git-wip-us.apache.org/repos/asf/incubator-tez/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tez/commit/3beb9f72
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tez/tree/3beb9f72
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tez/diff/3beb9f72
Branch: refs/heads/master
Commit: 3beb9f72b200bee8278b5e04de1ae804ea783dc1
Parents: 6674607
Author: Siddharth Seth <ss...@apache.org>
Authored: Wed Oct 23 22:03:47 2013 -0700
Committer: Siddharth Seth <ss...@apache.org>
Committed: Wed Oct 23 22:03:47 2013 -0700
----------------------------------------------------------------------
.../main/java/org/apache/tez/dag/api/DAG.java | 68 +++++++++++++++++---
.../org/apache/tez/dag/api/TestDAGVerify.java | 48 ++++++++++++++
2 files changed, 108 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/3beb9f72/tez-api/src/main/java/org/apache/tez/dag/api/DAG.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/dag/api/DAG.java b/tez-api/src/main/java/org/apache/tez/dag/api/DAG.java
index d5bbcd3..22a9a2e 100644
--- a/tez-api/src/main/java/org/apache/tez/dag/api/DAG.java
+++ b/tez-api/src/main/java/org/apache/tez/dag/api/DAG.java
@@ -152,27 +152,79 @@ public class DAG { // FIXME rename to Topology
throw new IllegalStateException("Invalid dag containing 0 vertices");
}
+ // check for valid vertices, duplicate vertex names,
+ // and prepare for cycle detection
+ Map<String, AnnotatedVertex> vertexMap = new HashMap<String, AnnotatedVertex>();
+ Map<Vertex, Set<String>> inboundVertexMap = new HashMap<Vertex, Set<String>>();
+ Map<Vertex, Set<String>> outboundVertexMap = new HashMap<Vertex, Set<String>>();
+ for (Vertex v : vertices.values()) {
+ if (vertexMap.containsKey(v.getVertexName())) {
+ throw new IllegalStateException("DAG contains multiple vertices"
+ + " with name: " + v.getVertexName());
+ }
+ vertexMap.put(v.getVertexName(), new AnnotatedVertex(v));
+ }
+
Map<Vertex, List<Edge>> edgeMap = new HashMap<Vertex, List<Edge>>();
for (Edge e : edges) {
+ // Construct structure for cycle detection
Vertex inputVertex = e.getInputVertex();
+ Vertex outputVertex = e.getOutputVertex();
List<Edge> edgeList = edgeMap.get(inputVertex);
if (edgeList == null) {
edgeList = new ArrayList<Edge>();
edgeMap.put(inputVertex, edgeList);
}
edgeList.add(e);
+
+ // Construct map for Input name verification
+ Set<String> inboundSet = inboundVertexMap.get(outputVertex);
+ if (inboundSet == null) {
+ inboundSet = new HashSet<String>();
+ inboundVertexMap.put(outputVertex, inboundSet);
+ }
+ inboundSet.add(inputVertex.getVertexName());
+
+ // Construct map for Output name verification
+ Set<String> outboundSet = outboundVertexMap.get(inputVertex);
+ if (outboundSet == null) {
+ outboundSet = new HashSet<String>();
+ outboundVertexMap.put(inputVertex, outboundSet);
+ }
+ outboundSet.add(outputVertex.getVertexName());
}
- // check for valid vertices, duplicate vertex names,
- // and prepare for cycle detection
- Map<String, AnnotatedVertex> vertexMap = new HashMap<String, AnnotatedVertex>();
- for (Vertex v : vertices.values()) {
- if (vertexMap.containsKey(v.getVertexName())) {
- throw new IllegalStateException("DAG contains multiple vertices"
- + " with name: " + v.getVertexName());
+ // Check for valid InputNames
+ for (Entry<Vertex, Set<String>> entry : inboundVertexMap.entrySet()) {
+ Vertex vertex = entry.getKey();
+ for (RootInputLeafOutput<InputDescriptor> input : vertex.getInputs()) {
+ if (entry.getValue().contains(input.getName())) {
+ throw new IllegalStateException("Vertex: "
+ + vertex.getVertexName()
+ + " contains an incoming vertex and Input with the same name: "
+ + input.getName());
+ }
+ }
+ }
+
+ // Check for valid OutputNames
+ for (Entry<Vertex, Set<String>> entry : outboundVertexMap.entrySet()) {
+ Vertex vertex = entry.getKey();
+ for (RootInputLeafOutput<OutputDescriptor> output : vertex.getOutputs()) {
+ if (entry.getValue().contains(output.getName())) {
+ throw new IllegalStateException("Vertex: "
+ + vertex.getVertexName()
+ + " contains an outgoing vertex and Output with the same name: "
+ + output.getName());
+ }
}
- vertexMap.put(v.getVertexName(), new AnnotatedVertex(v));
}
+
+
+ // Not checking for repeated input names / output names vertex names on the same vertex,
+ // since we only allow 1 at the moment.
+ // When additional inputs are supported, this can be chceked easily (and early)
+ // within the addInput / addOutput call itself.
detectCycles(edgeMap, vertexMap);
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/3beb9f72/tez-api/src/test/java/org/apache/tez/dag/api/TestDAGVerify.java
----------------------------------------------------------------------
diff --git a/tez-api/src/test/java/org/apache/tez/dag/api/TestDAGVerify.java b/tez-api/src/test/java/org/apache/tez/dag/api/TestDAGVerify.java
index fa773a2..14c510a 100644
--- a/tez-api/src/test/java/org/apache/tez/dag/api/TestDAGVerify.java
+++ b/tez-api/src/test/java/org/apache/tez/dag/api/TestDAGVerify.java
@@ -279,6 +279,54 @@ public class TestDAGVerify {
Assert.assertTrue(ex.getMessage().startsWith("Vertex v1 already defined"));
}
+ @Test (expected = IllegalStateException.class)
+ public void testInputAndInputVertexNameCollision() {
+ Vertex v1 = new Vertex("v1",
+ new ProcessorDescriptor("MapProcessor"),
+ dummyTaskCount, dummyTaskResource);
+ Vertex v2 = new Vertex("v2",
+ new ProcessorDescriptor("MapProcessor"),
+ dummyTaskCount, dummyTaskResource);
+
+ v2.addInput("v1", new InputDescriptor(), null);
+
+ Edge e1 = new Edge(v1, v2,
+ new EdgeProperty(DataMovementType.SCATTER_GATHER,
+ DataSourceType.PERSISTED, SchedulingType.SEQUENTIAL,
+ new OutputDescriptor("dummy output class"),
+ new InputDescriptor("dummy input class")));
+
+ DAG dag = new DAG("testDag");
+ dag.addVertex(v1);
+ dag.addVertex(v2);
+ dag.addEdge(e1);
+ dag.verify();
+ }
+
+ @Test (expected = IllegalStateException.class)
+ public void testOutputAndOutputVertexNameCollision() {
+ Vertex v1 = new Vertex("v1",
+ new ProcessorDescriptor("MapProcessor"),
+ dummyTaskCount, dummyTaskResource);
+ Vertex v2 = new Vertex("v2",
+ new ProcessorDescriptor("MapProcessor"),
+ dummyTaskCount, dummyTaskResource);
+
+ v1.addOutput("v2", new OutputDescriptor());
+
+ Edge e1 = new Edge(v1, v2,
+ new EdgeProperty(DataMovementType.SCATTER_GATHER,
+ DataSourceType.PERSISTED, SchedulingType.SEQUENTIAL,
+ new OutputDescriptor("dummy output class"),
+ new InputDescriptor("dummy input class")));
+
+ DAG dag = new DAG("testDag");
+ dag.addVertex(v1);
+ dag.addVertex(v2);
+ dag.addEdge(e1);
+ dag.verify();
+ }
+
// v1 v2
// | |
// v3