EmptyFlattenAsCreateFactory implements PTransformOverrideFactory<
        PCollectionList<T>, PCollection<T>, Flatten.PCollections<T>> { @Override public PTransformReplacement<PCollectionList<T>, PCollection<T>> getReplacementTransform( AppliedPTransform<PCollectionList<T>, PCollection<T>, PCollections<T>> transform) { checkArgument( transform.getInputs().isEmpty(), "Unexpected nonempty input %s for %s", transform.getInputs(), getClass().getSimpleName()); return PTransformReplacement.of( PCollectionList.empty(transform.getPipeline()), new CreateEmptyFromList<T>()); } private  EmptyFlattenAsCreateFactory(); static EmptyFlattenAsCreateFactory<T> instance(); @Override PTransformReplacement<PCollectionList<T>, PCollection<T>> getReplacementTransform(
      AppliedPTransform<PCollectionList<T>, PCollection<T>, PCollections<T>> transform); @Override Map<PValue, ReplacementOutput> mapOutputs(
      Map<TupleTag<?>, PValue> outputs, PCollection<T> newOutput);  }
@Test public void getInputEmptySucceeds() { PTransformReplacement<PCollectionList<Long>, PCollection<Long>> replacement = factory.getReplacementTransform( AppliedPTransform.of( "nonEmptyInput", Collections.emptyMap(), Collections.emptyMap(), Flatten.pCollections(), pipeline)); assertThat(replacement.getInput().getAll(), emptyIterable()); }
@Test public void getInputNonEmptyThrows() { PCollectionList<Long> nonEmpty = PCollectionList.of(pipeline.apply("unbounded", GenerateSequence.from(0))) .and(pipeline.apply("bounded", GenerateSequence.from(0).to(100))); thrown.expect(IllegalArgumentException.class); thrown.expectMessage(nonEmpty.expand().toString()); thrown.expectMessage(EmptyFlattenAsCreateFactory.class.getSimpleName()); factory.getReplacementTransform( AppliedPTransform.of( "nonEmptyInput", nonEmpty.expand(), Collections.emptyMap(), Flatten.pCollections(), pipeline)); }
@Test @Category(NeedsRunner.class) public void testOverride() { PCollectionList<Long> empty = PCollectionList.empty(pipeline); PCollection<Long> emptyFlattened = empty.apply( factory .getReplacementTransform( AppliedPTransform.of( "nonEmptyInput", Collections.emptyMap(), Collections.emptyMap(), Flatten.pCollections(), pipeline)) .getTransform()); PAssert.that(emptyFlattened).empty(); pipeline.run(); }