Logical vs Physical Plan

Tez exposes two planes, and confusing them is the root of most "why did my DAG do that" questions.

• Logical plan — what the application author writes: Vertexes, Edges, each edge carrying an EdgeProperty. Lives in tez-api. It says what connects to what and how data should move, but nothing about which task index reads from which.
• Physical plan — what the AM actually schedules: TaskImpl instances per vertex, per-edge routing decisions, container assignments. Lives in tez-dag. Mutable at runtime through VertexManager reconfiguration and EdgeManagerPlugin routing.

The bridge between them is the EdgeManagerPlugin: it turns "these two vertices are connected by a SCATTER_GATHER edge" into concrete answers to "how many physical inputs does destination task 3 have, and which source output does each carry."

After this chapter you can compute, for any DataMovementType, exactly how many physical inputs and outputs a task sees; read the built-in edge managers and explain their routing; and reason about what happens to routing when a VertexManager changes parallelism mid-flight. This chapter is the physical counterpart to ipo-abstractions.md, which covers the per-IO contracts, and it feeds directly into shuffle-sort.md, where those physical inputs become fetches.

The logical plane

ls tez-api/src/main/java/org/apache/tez/dag/api/ | grep -E "DAG|Vertex|Edge|EdgeProperty"

EdgeProperty — four orthogonal axes

grep -n "enum DataMovementType\|enum DataSourceType\|enum SchedulingType\|public static EdgeProperty create" \
  tez-api/src/main/java/org/apache/tez/dag/api/EdgeProperty.java

// tez-api, org.apache.tez.dag.api.EdgeProperty
public enum DataMovementType { ONE_TO_ONE, BROADCAST, SCATTER_GATHER, CUSTOM }
public enum DataSourceType   { PERSISTED, PERSISTED_RELIABLE, EPHEMERAL }
public enum SchedulingType   { SEQUENTIAL, CONCURRENT }

A logical edge says nothing about which destination task index reads from which source task index. That decision belongs entirely to the EdgeManagerPlugin.

The physical plane

When the AM initializes a DAG it builds, per logical vertex:

ls tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/ | \
  grep -E "VertexImpl|TaskImpl|TaskAttemptImpl|^Edge.java"

• VertexImpl — the runtime vertex state machine (see vertex-lifecycle.md).
• TaskImpl[] — one per parallelism slot.
• TaskAttemptImpl — one per attempt of each task.
• Edge — the runtime edge object holding a live EdgeManagerPlugin.

flowchart LR
  subgraph logical[Logical]
    LV1["Vertex A (parallelism 3)"]
    LV2["Vertex B (parallelism 2)"]
    LV1 -- "EdgeProperty SCATTER_GATHER" --> LV2
  end
  subgraph physical[Physical]
    A0[A.0] --> B0[B.0]
    A0 --> B1[B.1]
    A1[A.1] --> B0
    A1 --> B1
    A2[A.2] --> B0
    A2 --> B1
  end
  logical --> physical

Each source task produces one output partition per destination task; each destination task reads one physical input per source task. The EdgeManager decides which output partition goes to which input index.

The task-level view: TaskSpec

Before a task runs, the AM hands the container a TaskSpec — the physical plan, serialized down to one task. This is the concrete artifact the two planes meet in.

grep -n "private\|public TaskSpec(" \
  tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TaskSpec.java

// tez-runtime-internals, org.apache.tez.runtime.api.impl.TaskSpec
private TezTaskAttemptID taskAttemptId;
private String dagName;
private String vertexName;
private ProcessorDescriptor processorDescriptor;
private List<InputSpec>  inputSpecList;
private List<OutputSpec> outputSpecList;
private List<GroupInputSpec> groupInputSpecList;
private int vertexParallelism = -1;
private Configuration taskConf;

Each InputSpec carries the physical edge count — the number the framework passes to your AbstractLogicalInput(context, numPhysicalInputs) constructor:

// tez-runtime-internals, org.apache.tez.runtime.api.impl.InputSpec
private String sourceVertexName;
private InputDescriptor inputDescriptor;
private int physicalEdgeCount;      // <- becomes numPhysicalInputs

OutputSpec mirrors it with destinationVertexName and physicalEdgeCount. The physicalEdgeCount on an InputSpec is exactly EdgeManagerPlugin.getNumDestinationTaskPhysicalInputs(destTaskIndex). That is the seam: the AM asks the edge manager, stamps the answer into the InputSpec, ships the TaskSpec, and the runtime constructs your input with it.

EdgeManagerPlugin — the routing brain

grep -n "public abstract" tez-api/src/main/java/org/apache/tez/dag/api/EdgeManagerPlugin.java

// tez-api, org.apache.tez.dag.api.EdgeManagerPlugin
public abstract void initialize() throws Exception;
public abstract int  getNumDestinationTaskPhysicalInputs(int destinationTaskIndex) throws Exception;
public abstract int  getNumSourceTaskPhysicalOutputs(int sourceTaskIndex) throws Exception;
public abstract void routeDataMovementEventToDestination(DataMovementEvent event,
    int sourceTaskIndex, int sourceOutputIndex,
    Map<Integer, List<Integer>> destinationTaskAndInputIndices) throws Exception;
public abstract void routeInputSourceTaskFailedEventToDestination(int sourceTaskIndex,
    Map<Integer, List<Integer>> destinationTaskAndInputIndices) throws Exception;
public abstract int  getNumDestinationConsumerTasks(int sourceTaskIndex) throws Exception;
public abstract int  routeInputErrorEventToSource(InputReadErrorEvent event,
    int destinationTaskIndex, int destinationFailedInputIndex) throws Exception;

The plugin is constructed with an EdgeManagerPluginContext, its only window onto the DAG:

// tez-api, org.apache.tez.dag.api.EdgeManagerPluginContext
public UserPayload getUserPayload();
public String getSourceVertexName();
public String getDestinationVertexName();
public int    getSourceVertexNumTasks();
public int    getDestinationVertexNumTasks();
String        getVertexGroupName();

Notice what is not here: the plugin cannot see task states, locations, or the DAG. It is a pure function of the two vertices' task counts plus its own payload. That constraint is deliberate — it is what lets the AM call these methods thousands of times per second during event routing without locking the DAG.

EdgeManagerPluginOnDemand — the fast path

The map-filling routeDataMovementEventToDestination is the legacy API. Modern built-ins extend EdgeManagerPluginOnDemand, which returns compact EventRouteMetadata / CompositeEventRouteMetadata instead of mutating a map — the AM materializes per-destination events lazily, only when a destination task actually starts.

grep -n "public abstract\|class EventRouteMetadata\|class CompositeEventRouteMetadata" \
  tez-api/src/main/java/org/apache/tez/dag/api/EdgeManagerPluginOnDemand.java

// tez-api, org.apache.tez.dag.api.EdgeManagerPluginOnDemand
public abstract void prepareForRouting() throws Exception;
public abstract @Nullable EventRouteMetadata routeDataMovementEventToDestination(
    int sourceTaskIndex, int sourceOutputIndex, int destinationTaskIndex) throws Exception;
public abstract @Nullable CompositeEventRouteMetadata routeCompositeDataMovementEventToDestination(
    int sourceTaskIndex, int destinationTaskIndex) throws Exception;

Edge picks the routing mode at routingToBegin():

// tez-dag, org.apache.tez.dag.app.dag.impl.Edge.routingToBegin()
if (numDestTasks == 0) {
  routingNeeded = false;
} else if (numDestTasks < 0) {
  throw new TezUncheckedException("Internal error. Not expected to route events ... " +
      "until parallelism is determined ...");
}
if (edgeManager instanceof EdgeManagerPluginOnDemand) {
  onDemandRouting = true;
}
if (onDemandRouting) {
  ((EdgeManagerPluginOnDemand) edgeManager).prepareForRouting();
}

Note: numDestTasks < 0 throws. An edge cannot route until the destination vertex's parallelism is known. This is why auto-parallelism (below) must resolve before any event flows to the destination — the whole ordering is enforced right here.

The built-in edge managers

The three canonical routers live in tez-dag; the CUSTOM ones ship in tez-runtime-library. Do not assume they are all in one place:

find tez-dag/src/main/java -name "*EdgeManager*.java"
find tez-runtime-library/src/main/java -name "*EdgeManager*.java"

SCATTER_GATHER

sed -n '55,65p;100,165p' \
  tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/ScatterGatherEdgeManager.java

The counting methods are two lines each, and they are the whole story:

// tez-dag, org.apache.tez.dag.app.dag.impl.ScatterGatherEdgeManager
public int getNumDestinationTaskPhysicalInputs(int destinationTaskIndex) {
  return getContext().getSourceVertexNumTasks();       // = source parallelism
}
public int getNumSourceTaskPhysicalOutputs(int sourceTaskIndex) {
  return getContext().getDestinationVertexNumTasks();  // = dest parallelism
}

So for A (parallelism 3) → B (parallelism 2): every A task emits 2 partitions (one per B task); every B task has 3 physical inputs (one per A task). The on-demand routing collapses to index equality:

// tez-dag, ScatterGatherEdgeManager.routeDataMovementEventToDestination (on-demand)
public EventRouteMetadata routeDataMovementEventToDestination(
    int sourceTaskIndex, int sourceOutputIndex, int destinationTaskIndex) {
  if (sourceOutputIndex == destinationTaskIndex) {          // partition d -> task d
    return getOrCreateCommonRouteMeta().get(sourceTaskIndex);
  }
  return null;                                              // this output not for this dest
}

Invariant to memorize: numSourceOutputs == destParallelism and numDestInputs == srcParallelism. Every DataMovementEvent a B task receives corresponds to one A attempt's partition-d output.

ONE_TO_ONE

sed -n '45,90p' tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/OneToOneEdgeManager.java

// tez-dag, org.apache.tez.dag.app.dag.impl.OneToOneEdgeManager
public void routeDataMovementEventToDestination(DataMovementEvent event,
    int sourceTaskIndex, int sourceOutputIndex,
    Map<Integer, List<Integer>> destinationTaskAndInputIndices) {
  checkState();
  destinationTaskAndInputIndices.put(sourceTaskIndex, destinationInputIndices);
}

private void checkState() {
  Preconditions.checkState(getContext().getSourceVertexNumTasks()
      == getContext().getDestinationVertexNumTasks(),
      "1-1 source and destination task counts must match. ...");
}

Source i feeds destination i, one physical input each, getNumDestinationConsumerTasks returns 1. The checkState() precondition is the trap: if anything changes source or destination parallelism asymmetrically after the edge is live, this Preconditions.checkState throws. Never enable auto-parallelism on a vertex whose incoming edge is ONE_TO_ONE.

BROADCAST

sed -n '40,120p' tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/BroadcastEdgeManager.java

// tez-dag, org.apache.tez.dag.app.dag.impl.BroadcastEdgeManager
public int getNumDestinationTaskPhysicalInputs(int destinationTaskIndex) {
  return getContext().getSourceVertexNumTasks();   // every dest reads all sources
}
public void routeDataMovementEventToDestination(DataMovementEvent event,
    int sourceTaskIndex, int sourceOutputIndex,
    Map<Integer, List<Integer>> destinationTaskAndInputIndices) {
  for (int i = 0; i < getContext().getDestinationVertexNumTasks(); ++i) {
    // every source output is delivered to every destination task
  }
}

Each source emits a single logical output consumed by every destination task. Data volume amplification is srcParallelism × destParallelism — a 500-task broadcast source into a 500-task destination replicates each byte 500 times. Broadcasting a large vertex is an antipattern; keep broadcast sides small (dimension tables, not facts).

CUSTOM — Cartesian product

find tez-runtime-library/src/main/java -path "*cartesianproduct*EdgeManager*.java"

CartesianProductVertexManager chunks each source's outputs and the CartesianProductEdgeManagerReal projects a (sourceChunk, position) pair onto a 2-D grid of destination tasks. CUSTOM is the escape hatch by which Hive ships routing for unconventional joins that none of the three built-ins express.

The other two axes: durability and scheduling

DataMovementType gets all the attention, but the other EdgeProperty axes change behavior, not just routing, and they interact with the physical plan.

DataSourceType governs what happens to a source's output when the source attempt dies:

This is why a fetch-failure storm (see shuffle-sort.md) on a PERSISTED edge triggers source re-execution: the persisted output on the dead node is gone, so the only way to recover partition p is to re-run the source task that produced it.

SchedulingType decides whether the destination may start before the source finishes. SEQUENTIAL (the shuffle default) means destinations wait for source completion events; CONCURRENT lets both run together, which broadcast and pipelined shuffle require. A CONCURRENT edge changes the scheduler's readiness calculus for the whole vertex — see scheduler.md.

Merged inputs at the task level: GroupInputSpec

Recall TaskSpec also carries a groupInputSpecList. When several logical edges feed a MergedLogicalInput (see ipo-abstractions.md), the AM records that grouping in a GroupInputSpec so the runtime knows to construct the merged wrapper and route the constituent inputs' start() through it rather than auto-starting them individually. This is the physical-plan counterpart to a vertex-group union: three source vertices, three InputSpecs, one GroupInputSpec, one reader visible to the processor.

Runtime mutation: parallelism reconfiguration

A logical Vertex declares a target parallelism. The physical parallelism can still change — but only before the vertex schedules any task — via the VertexManager.

grep -n "reconfigureVertex\|setVertexParallelism\|vertexReconfigurationPlanned" \
  tez-api/src/main/java/org/apache/tez/dag/api/VertexManagerPluginContext.java

The modern API is reconfigureVertex, with three overloads:

// tez-api, org.apache.tez.dag.api.VertexManagerPluginContext
void reconfigureVertex(int parallelism, @Nullable VertexLocationHint locationHint,
    @Nullable Map<String, EdgeProperty> sourceEdgeProperties,
    @Nullable Map<String, InputSpecUpdate> rootInputSpecUpdate);

void reconfigureVertex(int parallelism, @Nullable VertexLocationHint locationHint,
    @Nullable Map<String, EdgeProperty> sourceEdgeProperties);

void reconfigureVertex(@Nullable Map<String, InputSpecUpdate> rootInputSpecUpdate,
    @Nullable VertexLocationHint locationHint, int parallelism);

A reconfiguration does three things atomically inside VertexImpl: resizes the TaskImpl[], re-installs EdgeManagerPlugin instances on incoming edges (via sourceEdgeProperties), and updates location hints. The state-machine guard that makes it safe:

sed -n '1872,1900p' tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java

// tez-dag, org.apache.tez.dag.app.dag.impl.VertexImpl.setParallelismWrapper()
if (!tasksNotYetScheduled) {
  String msg = "setParallelism cannot be called after scheduling tasks. Vertex: "
      + getLogIdentifier();
  throw new TezUncheckedException(msg);
}
if (fromVertexManager && canInitVertex()) {
  Preconditions.checkState(vertexToBeReconfiguredByManager,
      "Vertex is fully configured but still the reconfiguration API has been called. "
    + "VertexManager must notify the framework using "
    + "context.vertexReconfigurationPlanned() before re-configuring the vertex. ...");
}

Two rules fall straight out of this code:

1. Reconfiguration is illegal once any task is scheduled (!tasksNotYetScheduled throws).
2. You must announce intent first. A VertexManager that will reconfigure a fully-defined vertex must call context.vertexReconfigurationPlanned() before reconfigureVertex, or the checkState fires. This is a common custom-VertexManager bug.

Worked example: ShuffleVertexManager auto-parallelism

ShuffleVertexManager is the built-in that shrinks an over-provisioned reduce vertex based on real upstream data size.

grep -n "DESIRED_TASK_INPUT_SIZE\|ENABLE_AUTO_PARALLEL\|MIN_TASK_PARALLELISM\|MIN_SRC_FRACTION\|MAX_SRC_FRACTION" \
  tez-runtime-library/src/main/java/org/apache/tez/dag/library/vertexmanager/ShuffleVertexManager.java

The sequence:

1. Vertex R declared with parallelism 100 (a pessimistic upper bound).
2. Each upstream task's output close() emits a VertexManagerEvent carrying per-partition byte counts (see ipo-abstractions.md).
3. ShuffleVertexManager.onVertexManagerEventReceived accumulates the totals.
4. After min-src-fraction of sources report, it estimates target = ceil(totalBytes / desiredTaskInputSize), clamped to [min-task-parallelism, originalParallelism].
5. It calls reconfigureVertex(target, null, updatedEdgeProps) — say 100 → 17.
6. VertexImpl resizes TaskImpl[] and rebuilds the incoming SCATTER_GATHER edge's manager so 100-partition outputs route into 17 destinations (each destination now reads a range of the original partitions).

sequenceDiagram
  participant SRC as Upstream tasks
  participant VM as ShuffleVertexManager
  participant V as VertexImpl (R)
  participant E as Incoming Edge
  SRC-->>VM: VertexManagerEvent (per-partition byte stats)
  VM->>VM: accumulate; wait for min-src-fraction
  VM->>V: reconfigureVertex(17, hint, edgeProps)
  V->>V: resize TaskImpl[] 100 -> 17  (tasksNotYetScheduled == true)
  V->>E: setEdgeProperty(new EdgeManager)
  Note over E: now routes 100 partitions -> 17 dests
  V->>V: schedule 17 tasks

Reading exercise

# When is the EdgeManagerPlugin instantiated / re-installed?
grep -n "createEdgeManager\|edgeManager =\|setEdgeProperty" \
  tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/Edge.java

# Factory methods on EdgeProperty and which require an EdgeManagerPluginDescriptor
grep -n "public static EdgeProperty create" \
  tez-api/src/main/java/org/apache/tez/dag/api/EdgeProperty.java

# The parallelism guard, verbatim
sed -n '1872,1900p' tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java

# Tests that pin each router's behavior
grep -rln "ScatterGatherEdgeManager\|BroadcastEdgeManager\|OneToOneEdgeManager" tez-dag/src/test

Answer:

1. For SCATTER_GATHER, A(3) → B(2): how many partitions does each A task emit, and how many physical inputs does each B task have? Cite the two getNum... methods.
2. For ONE_TO_ONE, what exactly throws if the upstream auto-parallelizes 100 → 17 after the destination is initialized? Quote the precondition.
3. For BROADCAST, express the data amplification in terms of source and destination parallelism.
4. Which EdgeManagerPlugin methods are called once per edge, and which per destination-task init? Use EdgeManagerPluginOnDemand.prepareForRouting vs routeDataMovementEventToDestination to answer.
5. Trace where an InputSpec's physicalEdgeCount comes from, from getNumDestinationTaskPhysicalInputs to the constructor of your AbstractLogicalInput.

Then answer in prose:

• Why must the destination vertex's parallelism be resolved before any event routes to it? Cite the numDestTasks < 0 branch in Edge.routingToBegin.
• What must a custom VertexManager call before reconfigureVertex on a fully-defined vertex, and what happens if it forgets?

Common bugs and symptoms

Validation: prove you understand this

1. Given A(parallelism 4) SCATTER_GATHER→ B(parallelism 3), compute the number of DataMovementEvents that B.1 receives. Show the arithmetic from the two getNum... methods.
2. In one sentence each: when is an EdgeManagerPlugin re-instantiated, and when does it survive a reconfiguration? Cite Edge.setEdgeProperty.
3. Write a one-paragraph rejection of "let's just use BROADCAST for our 500-task lookup vertex," citing the concrete amplification factor.
4. Quote the exact line in VertexImpl.setParallelismWrapper (from your own grep -n, not a memorized number) that rejects reconfiguration after scheduling.
5. Sketch a CUSTOM EdgeManagerPlugin for a range-partitioned merge where source task i emits keys in [i*R, (i+1)*R) and the destination is K tasks with K possibly ≠ source parallelism. Define getNumDestinationTaskPhysicalInputs, getNumSourceTaskPhysicalOutputs, and the on-demand routing rule in code.
6. Trace one VertexManagerEvent from an upstream close() (see ipo-abstractions.md) to a reconfigureVertex call, naming every class it passes through.