cloudera/crunch — pull request #25 by tzolov (2012-05-11 21:33:12+00:00)
Title: Fix NPE thrown on write or materialize of Union collections

An attempt to write or materialize UnionCollection throws NPE!
Direct union.write()/union.materialize() calls as well as pipeline.write(union)/pipeline.materialize(union) lead to the same problem as illustrated below:

```java
PCollection one = mrPipeline.readTextFile(...);
PCollection other = mrPipeline.readTextFile(...);

PCollection union = other.union(one);
union.write(To.textFile(...));
//or union.materialize();

mrPipeline.done();
```
```java
java.lang.NullPointerException
at com.cloudera.crunch.impl.mr.plan.MSCRPlanner.plan(MSCRPlanner.java:154)
at com.cloudera.crunch.impl.mr.MRPipeline.run(MRPipeline.java:94)
```
I found that the MRPipeline.writeTextFile() implementation injects an extra IdentityFn collection between the input collection and the target. This seems to solve the problem so I have applied this technique to the MRPipeline.write() and MRPipeline.materialize(). This also solved the union.write()/union.materialize() as they use to the pipeline internally.

The UnionCollectionTest is provided to validate the solution across all type families and pipelines.

> Note: To make it consistent with the existing MRPipeline.union() implementation I had to reverse the inclusion order in the MemPipeline.union() implementation

Although the suggested solution resolves the NPE I am not sure if this is the right or intended solution? As an alternative an IllegalStateException can be thrown to indicate that UnionCollection should not be serialized directly.