Lab 3.3 — Build It: Multi-Input Union DAG

Background

A vertex with two upstream sources is the structural heart of every join, union, and co-group in Tez. Until now you have read single-input processors (OrderedWordCount's SumProcessor in Lab 3.2). This lab makes you build the smallest possible program that has a multi-input vertex: two source vertices, each emitting a stream of integers, both feeding one middle vertex that unions the streams and forwards everything to a terminal sink. The sink sums every value and publishes the total as a Tez counter, and the driver asserts the counter equals a known ground truth.

EvenSource(1) ──▶┐
                 ├─▶ UnionProcessor(1) ──▶ UnionSink(1)  ──▶  counter TotalSum
OddSource(1)  ──▶┘

Even source emits 0, 2, 4, …, 98; odd emits 1, 3, 5, …, 99. The union forwards all 100 values; the sink sums them. Expected result: TotalSum=4950 PASS.

The deliverable is a real Maven module — book/projects/level-3-multi-input — that you build, run, break on purpose, and extend. The single most important thing you will learn, and the thing this lab forces you to get right against real Tez source, is how a processor retrieves more than one input: getInputs() is a Map<String, LogicalInput> keyed by the upstream (source) vertex name. You proved this in Lab 3.1 (runInputMap.put(inputSpec.getSourceVertexName(), input)) and saw it used in Lab 3.2 (getInputs().get(TOKENIZER)). Now you depend on it.

Companion deep dives: ipo-abstractions.md, dag-app-master.md; the Level 3 overview frames the whole level.

Lab type: Build It — real Maven project, compilable Java, run + break + fix cycle
Estimated time: 90–120 min
Maven module: book/projects/level-3-multi-input
Main class: org.apache.tez.learning.l3.MultiInputDAG

Why This Lab Matters for Contributors

Multi-input vertices are where correctness bugs bite hardest: an input retrieved by the wrong key returns null; a reader drained in the wrong order deadlocks; a join that assumes left-before-right corrupts results. Committers reviewing join/union patches must be able to reason about exactly which LogicalInput a processor pulled and why. Building this tiny union from an empty pom.xml gives you that reasoning in your hands — and the "Break It" experiments below reproduce the three most common real failures so you recognize them in a bug report on sight.

There is also a sharper lesson buried here. The naive mental model — "I'll name my edges and look them up by edge name" — is wrong for Tez's public API. There is no per-edge name on the public Edge class; the input map key is the source vertex name. This lab makes you verify that against the source and wire your processor accordingly. Getting this right is the difference between code that compiles-and-runs and code that throws NullPointerException at the first getInputs().get(...).

Prerequisites

Apache Tez cloned and installed to your local Maven repo (mvn -DskipTests install), so the companion project can resolve tez-api, tez-common, tez-runtime-library.
Completed Lab 3.1 and Lab 3.2.
JDK and Maven on your path; ability to run a fat JAR.
Your Tez checkout for verification:

export TEZ=~/src/tez

Step 1 — Match the Tez version

Open book/projects/pom.xml. By default the companion projects pin the latest released Tez (<tez.version>0.10.5</tez.version>), which resolves from Maven Central — so mvn test builds them out of the box with no local Tez build required. To compile against a Tez you built from source (Lab 1.1 — e.g. master, currently 1.0.0-SNAPSHOT), activate the local-tez profile and set the version to match your build:

# Default (released Tez from Maven Central):
mvn -q -pl level-3-multi-input -am test
# Against your locally-installed source build:
mvn -q -Plocal-tez -Dtez.version=1.0.0-SNAPSHOT -pl level-3-multi-input -am test

cd "$TEZ"
git log --oneline -1
mvn help:evaluate -Dexpression=project.version -q -DforceStdout ; echo

If the printed version differs from <tez.version> in book/projects/pom.xml, update the POM before continuing, or the module will not resolve its dependencies.

Step 2 — Compile and run the unit tests

cd /path/to/opensource-engineer-and-contributor/book/projects
mvn -pl level-3-multi-input test

Expected:

Tests run: 12, Failures: 0, Errors: 0, Skipped: 0

These are pure-logic tests in TestMultiInputProcessors.java — they check arithmetic constants, the expected sum, edge-name/counter constants, and (the good one) a brute-force boolean[100] proof that even + odd cover 0..99 with no overlap and no gap. Read every test before moving on.

Questions

#	Question
1	`testEvenAndOddRangesNoOverlapNoGap` simulates both sources with a `boolean[]`. Why is that more rigorous than just checking the two counts sum to 100?
2	`testEdgeNameConstants` pins string literals. What runtime bug appears if a developer renames a constant in the processor but not the DAG wiring? (Hint: a `null` from `getInputs().get(...)`.)
3	`testExpectedSum` hardcodes `4950L`. Could you make it fail by changing only `EvenNumberSource.COUNT`? What else would have to change to keep it PASS?

Step 3 — Build the fat JAR and run the DAG

mvn -pl level-3-multi-input package -q
java -jar level-3-multi-input/target/level-3-multi-input-1.0-SNAPSHOT-jar-with-dependencies.jar

Expected final line:

[MultiInputDAG] TotalSum=4950  expected=4950  PASS

If you see FAIL, note the actual counter value before the debugging exercises. The program runs in local mode (no YARN) — confirm the three settings the driver applies in MultiInputDAG.main:

tezConf.setBoolean(TezConfiguration.TEZ_LOCAL_MODE, true);
tezConf.set("fs.defaultFS", "file:///");
tezConf.setBoolean(TezConfiguration.TEZ_LOCAL_MODE_WITHOUT_NETWORK, true);

These are exactly the local-mode settings from Lab 3.1's breakpoint recipe — which means you can attach a debugger to this jar and watch the whole submission cascade.

Step 4 — Read every source file

Work through each file in level-3-multi-input/src/main/java/org/apache/tez/learning/l3/. Read them against the real Tez contracts you quoted in Lab 3.2.

`EvenNumberSource.java` (and its twin `OddNumberSource.java`)

A source is a processor with no logical input and one output. Its run():

OrderedPartitionedKVOutput output =
    (OrderedPartitionedKVOutput) getOutputs().values().iterator().next();
output.start();
KeyValueWriter writer = output.getWriter();
IntWritable key = new IntWritable();
IntWritable value = new IntWritable();
for (int i = 0; i < COUNT; i++) {
    int n = i * 2;      // odd source uses i * 2 + 1
    key.set(n); value.set(n);
    writer.write(key, value);
}

#	Question
1	It extends `AbstractLogicalIOProcessor`. What four methods must it therefore implement, and which one does the real work? (Cross-check the contract from Lab 3.2.)
2	Why does `run()` call `output.start()` before `getWriter()`? What exception results if you skip it? (You will trigger this in Break It — Experiment B.)
3	The output is fetched via `getOutputs().values().iterator().next()`. Since this vertex has exactly one output that is safe — but what would break if it had two, and how does the by-name idiom from Lab 3.2 fix it?
4	`key` and `value` are declared once, outside the loop, and reused via `set()`. What allocation cost would moving them inside the loop add on a source that emits millions of records?

`MultiInputUnionProcessor.java` — the multi-input vertex

This is the class the whole lab exists for. It has two logical inputs and one output. Read how it obtains them — and note the by-name lookup, because verifying that key is the core task of this step:

Map<String, LogicalInput> inputs   = getInputs();
Map<String, LogicalOutput> outputs = getOutputs();
OrderedGroupedKVInput evenInput = (OrderedGroupedKVInput) inputs.get(EVEN_EDGE);
OrderedGroupedKVInput oddInput  = (OrderedGroupedKVInput) inputs.get(ODD_EDGE);
evenInput.start();
oddInput.start();
OrderedPartitionedKVOutput output =
    (OrderedPartitionedKVOutput) outputs.values().iterator().next();
output.start();
KeyValueWriter writer = output.getWriter();
// drain even, then odd:
KeyValueReader evenReader = evenInput.getReader();
while (evenReader.next()) writer.write(evenReader.getCurrentKey(), evenReader.getCurrentValue());
KeyValueReader oddReader = oddInput.getReader();
while (oddReader.next()) writer.write(oddReader.getCurrentKey(), oddReader.getCurrentValue());

VERIFY THE KEY — this is the critical step of the lab. The processor looks up its inputs by the constants EVEN_EDGE / ODD_EDGE. For that lookup to return a non-null LogicalInput, those constants must equal the source vertex names, because Tez keys the input map by source vertex name — you verified this in Lab 3.1:
grep -n "runInputMap.put" \
  "$TEZ"/tez-runtime-internals/src/main/java/org/apache/tez/runtime/LogicalIOProcessorRuntimeTask.java
# -> runInputMap.put(inputSpec.getSourceVertexName(), input);
and again in the shipped examples:
grep -n "getInputs().get" "$TEZ"/tez-examples/src/main/java/org/apache/tez/examples/CartesianProduct.java
# -> getInputs().get(VERTEX1), getInputs().get(VERTEX2), ...  (VERTEXn are vertex names)
There is no per-edge name on the public Edge class — confirm it yourself:
grep -n "public " "$TEZ"/tez-api/src/main/java/org/apache/tez/dag/api/Edge.java
# create, getInputVertex, getOutputVertex, getEdgeProperty, getId — and nothing that names an edge
So the reliable, verified way to retrieve two inputs is to look them up by the two source vertex names. When you wire the DAG in Step 4's next file, make the source vertices' names equal the constants the processor looks up (i.e. name the sources "even-edge" and "odd-edge", or set the constants to the source vertex names). Getting these to agree is exactly what testEdgeNameConstants guards.

#	Question
1	Trace where `EVEN_EDGE` / `ODD_EDGE` must appear in `MultiInputDAG.buildDAG()` for `inputs.get(EVEN_EDGE)` to be non-null. Which name on which vertex must match?
2	Both inputs are `start()`ed before either reader is obtained. Could you instead start-even → read-even → start-odd → read-odd? What does `Input.start()`'s "non-blocking, may be called by the framework" contract (Lab 3.2) say about interleaving?
3	The even input is fully drained before the odd reader is even obtained. Is there a scenario where an odd record must be read before all even records? Why does draining sequentially still produce the correct total here?
4	The processor forwards records unchanged. What single change to `run()` would make it emit only distinct values across both sources?

`MultiInputDAG.java` — the wiring

Read buildDAG(). It creates four vertices and three edges, all on a shared OrderedPartitionedKVEdgeConfig (SCATTER_GATHER, IntWritable/IntWritable, HashPartitioner):

OrderedPartitionedKVEdgeConfig edgeCfg = OrderedPartitionedKVEdgeConfig
    .newBuilder(IntWritable.class.getName(), IntWritable.class.getName(),
        HashPartitioner.class.getName())
    .build();
Vertex even  = Vertex.create("EvenSource",   ProcessorDescriptor.create(EvenNumberSource.class.getName()), 1);
Vertex odd   = Vertex.create("OddSource",    ProcessorDescriptor.create(OddNumberSource.class.getName()), 1);
Vertex union = Vertex.create("UnionProcessor", ProcessorDescriptor.create(MultiInputUnionProcessor.class.getName()), 1);
Vertex sink  = Vertex.create("UnionSink",    ProcessorDescriptor.create(UnionSinkProcessor.class.getName()), 1);
Edge evenEdge = Edge.create(even, union, edgeCfg.createDefaultEdgeProperty());
Edge oddEdge  = Edge.create(odd,  union, edgeCfg.createDefaultEdgeProperty());
Edge sinkEdge = Edge.create(union, sink,  edgeCfg.createDefaultEdgeProperty());
return DAG.create("MultiInputUnionDAG")
    .addVertex(even).addVertex(odd).addVertex(union).addVertex(sink)
    .addEdge(evenEdge).addEdge(oddEdge).addEdge(sinkEdge);

Reconcile the names. For the union processor's inputs.get(EVEN_EDGE) / inputs.get(ODD_EDGE) to resolve, the source vertex names must equal those constants. If your constants are "even-edge"/"odd-edge", name the source vertices "even-edge"/"odd-edge"; if your sources are named "EvenSource"/"OddSource", set the constants to those. Either way, run testEdgeNameConstants and the DAG together — they only pass when the names agree. This is the hands-on payoff of the "keyed by source vertex name" rule.

#	Question
1	Both source→union edges use `edgeCfg.createDefaultEdgeProperty()`. Could the two sources use different edge configs? When would that be necessary (hint: different key/value types on a join)?
2	The DAG has 4 vertices. Draw the dependency graph. Which vertices can run in parallel, and which must wait? (Recall the `numInitedSourceVertices == sourceVertices.size()` gate from Lab 3.1.)
3	`waitForCompletion(EnumSet.of(StatusGetOpts.GET_COUNTERS))` — what does `GET_COUNTERS` do, and what would `status.getDAGCounters()` return if you omitted it?
4	The driver reads back `UnionSinkProcessor.COUNTER_GROUP` / `COUNTER_NAME`. Where does the sink increment that counter, and why is a counter (not stdout) the right way to assert correctness across JVMs?

`UnionSinkProcessor.java` — the assertion point

The sink has one input, no output, and publishes the sum as a counter via getContext().getCounters().findCounter(GROUP, NAME).increment(sum). Confirm it uses a KeyValueReader (singular) and that expectedSum() computes 0+1+…+99 = 4950 — the ground truth MultiInputDAG asserts against.

Step 5 — Break It: three experiments

Perform each, observe the failure, then revert before the next.

Experiment A — Swap the input names

In MultiInputUnionProcessor.run() swap which constant reads which reader (or swap the source vertex names in buildDAG). Rebuild and run.

Does the DAG succeed or fail? Is the sum still 4950?
Explain: because both sources carry the same (IntWritable, IntWritable) schema and the sink sums everything, swapping which stream is "even" vs "odd" doesn't change the total. Now argue why the same swap in a join (where left and right inputs have different schemas) would corrupt results or throw ClassCastException.

Experiment B — Remove a `start()` call

In MultiInputUnionProcessor.run() delete evenInput.start();. Rebuild and run.

What exception is thrown, and at which call? (It surfaces at evenInput.getReader().)

Find the guard in Tez source that enforces "start before read." Search the ordered grouped input and its base for the state check:

grep -rn "start\|isStarted\|IllegalStateException" \
  "$TEZ"/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/OrderedGroupedKVInput.java | head

Connect this to the IPO lifecycle you documented in Lab 3.2: reader/writer access before start() is illegal by contract.

Experiment C — Make one source emit duplicate keys

In EvenNumberSource.run() change int n = i * 2; to int n = 0; (every write uses key 0, value 0). Rebuild and run.

What is the counter value now, and does the DAG PASS or FAIL?
The even source now contributes 0 to the sum fifty times (total 0), so TotalSum drops to 1+3+…+99 = 2500. Explain what this reveals: the union/sink does not de-duplicate; every physical record flows through. On a SCATTER_GATHER ordered edge, duplicate keys are grouped for a grouped reader but here the sink reads them all with a KeyValueReader.

Step 6 — Extend It

Two extensions, each a real edge-config change verified against source.

Extension 1 — Add a third source (`MultiSource`)

Add a ThirdNumberSource (say, emitting the multiples of… your choice — just keep the ground truth computable), create it as a fourth source vertex, add a third source→union edge, and add a third constant + reader-drain block to MultiInputUnionProcessor. The map now has three entries.

Update UnionSinkProcessor.expectedSum() (or the driver's expected value) to the new ground truth.
Confirm the union vertex still initializes only after all three sources init — the numInitedSourceVertices == sourceVertices.size() gate from Lab 3.1 now waits for three.

Extension 2 — Make one side BROADCAST

SCATTER_GATHER sends each key to one downstream task by partition. Broadcast sends every record to every downstream task. Switch the odd source's edge to a broadcast edge using the real unordered config — verified API:

grep -n "public static Builder newBuilder\|createDefaultBroadcastEdgeProperty" \
  "$TEZ"/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedKVEdgeConfig.java

You will find UnorderedKVEdgeConfig.newBuilder(keyClass, valueClass) (two args — no partitioner, since broadcast doesn't partition) and createDefaultBroadcastEdgeProperty(). Rewire the odd edge:

UnorderedKVEdgeConfig bcastCfg = UnorderedKVEdgeConfig
    .newBuilder(IntWritable.class.getName(), IntWritable.class.getName())
    .build();
Edge oddEdge = Edge.create(odd, union, bcastCfg.createDefaultBroadcastEdgeProperty());

The broadcast input is now an unordered input, so in MultiInputUnionProcessor cast that side to the unordered input type and read it with a KeyValueReader (not OrderedGroupedKVInput). Confirm the class the broadcast edge wires in:
```
grep -n "UnorderedKVInput\|UnorderedKVOutput\|createDefaultBroadcastEdgeProperty" \
  "$TEZ"/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedKVEdgeConfig.java
```
Because the union vertex has parallelism 1, broadcast delivers the same records it would have gotten under scatter-gather, so the sum is unchanged — but if you raise the union parallelism to 2, a broadcast side is duplicated to both tasks while a scatter-gather side is split. Predict the new total, then verify. This is the exact semantic that makes broadcast the right choice for the small side of a hash join (each join task needs the whole small side).

Step 7 — Connect to real Tez

For each class the project uses, locate its source file in your Tez clone and record the path.

Class used in this project	Tez source file (relative to repo root)
`AbstractLogicalIOProcessor`	`tez-api/.../runtime/api/AbstractLogicalIOProcessor.java`
`OrderedPartitionedKVOutput`
`OrderedGroupedKVInput`
`OrderedPartitionedKVEdgeConfig`
`UnorderedKVEdgeConfig`
`HashPartitioner`
`Edge` / `EdgeProperty`

Then open a shipped multi-input example and compare:

grep -n "getInputs().get" "$TEZ"/tez-examples/src/main/java/org/apache/tez/examples/CartesianProduct.java
grep -n "getInputs().get\|hashSide\|streamingSide" \
  "$TEZ"/tez-examples/src/main/java/org/apache/tez/examples/HashJoinExample.java

Identify the union-/join-like vertex that receives edges from multiple sources.
How does its processor retrieve each input — by what key?
Compare to MultiInputUnionProcessor: what is the same (by-name lookup, per-input start), what is different (join semantics, broadcast side)?

Deliverables

mvn -pl level-3-multi-input test green; mvn ... package produces the fat JAR.
The JAR run prints TotalSum=4950 expected=4950 PASS.
Answers to every Step 4 question, grounded in the source you read.
Written proof (a grep result) that getInputs() is keyed by source vertex name and that the public Edge class has no per-edge name method.
Break It: the exception from Experiment B and the Tez guard that threw it; the altered totals from Experiments A and C.
Extend It: a working third source or a broadcast odd edge, with the recomputed ground truth and a note on how parallelism > 1 changes a broadcast side.
The completed Step 7 source-connection table.

Troubleshooting

Symptom	Likely cause	What to check
`NullPointerException` at `inputs.get(EVEN_EDGE)`	Constant does not equal a source vertex name	The input map is keyed by source vertex name; make the constant and the `Vertex.create("...")` name agree
Module won't resolve `tez-api`	Version mismatch	`<tez.version>` in `book/projects/pom.xml` must match your installed Tez (`mvn help:evaluate -Dexpression=project.version`)
`IllegalStateException` at `getReader()`	Missing `start()`	Every input/output must be `start()`ed before its reader/writer is fetched (Lab 3.2 lifecycle)
`ClassCastException` casting the broadcast input	Broadcast edge wires an unordered input	Cast the broadcast side to the unordered input type + `KeyValueReader`, not `OrderedGroupedKVInput`
DAG hangs after `Running DAG:`	Union vertex waiting on an un-inited source	The `numInitedSourceVertices == sourceVertices.size()` gate; check every source vertex actually initialized
`TotalSum` wrong but PASS logic didn't trip	Ground truth not updated after an extension	Update `expectedSum()`/the driver's expected value to match your new sources
`jar-with-dependencies` not found	Ran before `package`	`mvn -pl level-3-multi-input package` builds the assembly; run the exact jar name printed

Stretch Goals

Interleave the readers. Rewrite run() to alternate reading one record from each input. Does the total change? What does this tell you about ordering guarantees across two inputs?
MergedLogicalInput. Tez can present several physical inputs as one merged logical input. Read the API and a merged input, and sketch how you'd replace the two-reader drain with a single merged reader:
```
find "$TEZ" -name "MergedLogicalInput.java" -o -name "ConcatenatedMergedKeyValuesInput.java" | grep -v target
```
What does ConcatenatedMergedKeyValuesInput give you that two separate getReader() calls do not?
A FilterUnionProcessor. Extend AbstractLogicalIOProcessor, take a threshold from UserPayload (default 50), forward only values >= threshold, and increment a UnionPipeline/FilteredCount counter per dropped record. Wire it in place of the union processor:
```
ProcessorDescriptor.create(FilterUnionProcessor.class.getName())
    .setUserPayload(UserPayload.create(ByteBuffer.wrap("threshold=50".getBytes())));
```
With threshold 50, the surviving sum is 50+51+…+99 = 3725 and FilteredCount = 50. Assert both.
JIRA research. On issues.apache.org/jira, search project = TEZ AND text ~ "multi-input" AND resolution = Fixed. Find one resolved issue touching multiple inputs to one vertex: what was the bug, which class changed, was a test added, and what does it assert?

Validation / Self-check

Answer without looking back:

What is the type and key of the map returned by getInputs()? Where in the Tez runtime is that key assigned, and what did the grep prove about per-edge names?
In MultiInputUnionProcessor, why must the lookup constants equal the source vertex names for the DAG to run? What symptom appears when they don't?
Why does removing evenInput.start() throw, and at which call does the exception actually surface? Name the contract it violates.
In the four-vertex DAG, which vertices can run in parallel and which must wait? What gate makes the union vertex wait for both sources?
After Experiment C (even source emits key 0 fifty times), what is TotalSum and why? What does that reveal about de-duplication in this pipeline?
Switching the odd edge to broadcast leaves the total unchanged at union parallelism 1 but changes it at parallelism 2. Explain, and connect it to why a hash join broadcasts its small side.
Which two EdgeConfig classes and which two factory methods (createDefault…) did you use, and how do their argument lists differ (partitioner vs none)?

When you can answer all seven, the JAR prints PASS, and one extension runs green with a recomputed ground truth, you have completed Lab 3.3 — and Level 3. Return to the Level 3 overview.

Open-Source Engineer & Contributor