Major Description
I'm working on using the AM event mechanism to dynamically prune partitions at DAG runtime for certain queries. The query is:
select count
from srcpart join srcpart_double_hour on (srcpart.hr*2 = srcpart_double_hour.hr) where srcpart_double_hour.hour = 11;
This will result in two vertices connected through a broadcast edge. The vertex prepares two things: The list of partition keys (hr) that are being sent to the AM for dynamic pruning and the records to be used in the hash join.
The second vertex will block until all events are received (initializer) then it will load and process the hash join.
It's possible for queries like this to result in zero splits on the second vertex (i.e.: no matching rows for the join)
The exception I get when this is run is:
org.apache.tez.dag.api.TezUncheckedException: Event must be routed. sourceVertex=vertex_1408686217936_0003_3_00 srcIndex = 0 destAttemptId=vertex_1408686217936_0003_3_01 edgeManager=org.apache.tez.dag.app.dag.impl.BroadcastEdgeManager Ev\
ent type=DATA_MOVEMENT_EVENT
at org.apache.tez.dag.app.dag.impl.Edge.sendTezEventToDestinationTasks(Edge.java:371)
at org.apache.tez.dag.app.dag.impl.VertexImpl$RouteEventTransition.transition(VertexImpl.java:3372)
at org.apache.tez.dag.app.dag.impl.VertexImpl.scheduleTasks(VertexImpl.java:1088)
at org.apache.tez.dag.app.dag.impl.VertexManager$VertexManagerPluginContextImpl.scheduleVertexTasks(VertexManager.java:111)
at org.apache.tez.dag.app.dag.impl.ImmediateStartVertexManager.onVertexStarted(ImmediateStartVertexManager.java:49)
at org.apache.tez.dag.app.dag.impl.VertexManager.onVertexStarted(VertexManager.java:244)
at org.apache.tez.dag.app.dag.impl.VertexImpl.startVertex(VertexImpl.java:2923)
at org.apache.tez.dag.app.dag.impl.VertexImpl.access$5900(VertexImpl.java:169)
at org.apache.tez.dag.app.dag.impl.VertexImpl$StartTransition.transition(VertexImpl.java:2914)
at org.apache.tez.dag.app.dag.impl.VertexImpl$StartTransition.transition(VertexImpl.java:2906)
at org.apache.hadoop.yarn.state.StateMachineFactory$MultipleInternalArc.doTransition(StateMachineFactory.java:385)
at org.apache.hadoop.yarn.state.StateMachineFactory.doTransition(StateMachineFactory.java:302)
at org.apache.hadoop.yarn.state.StateMachineFactory.access$300(StateMachineFactory.java:46)
at org.apache.hadoop.yarn.state.StateMachineFactory$InternalStateMachine.doTransition(StateMachineFactory.java:448)
at org.apache.tez.dag.app.dag.impl.VertexImpl.handle(VertexImpl.java:1355)
at org.apache.tez.dag.app.dag.impl.VertexImpl.handle(VertexImpl.java:168)
at org.apache.tez.dag.app.DAGAppMaster$VertexEventDispatcher.handle(DAGAppMaster.java:1650)
at org.apache.tez.dag.app.DAGAppMaster$VertexEventDispatcher.handle(DAGAppMaster.java:1636)
at org.apache.hadoop.yarn.event.AsyncDispatcher.dispatch(AsyncDispatcher.java:173)
at org.apache.hadoop.yarn.event.AsyncDispatcher$1.run(AsyncDispatcher.java:106)
at java.lang.Thread.run(Thread.java:695)