ViewOverrideFactory implements PTransformOverrideFactory<
        PCollection<ElemT>,
        PCollection<ElemT>,
        PTransform<PCollection<ElemT>, PCollection<ElemT>>> { @Override public PTransformReplacement<PCollection<ElemT>, PCollection<ElemT>> getReplacementTransform( AppliedPTransform< PCollection<ElemT>, PCollection<ElemT>, PTransform<PCollection<ElemT>, PCollection<ElemT>>> transform) { PCollectionView<ViewT> view; try { view = CreatePCollectionViewTranslation.getView(transform); } catch (IOException exc) { throw new RuntimeException( String.format( "Could not extract %s from transform %s", PCollectionView.class.getSimpleName(), transform), exc); } return PTransformReplacement.of( PTransformReplacements.getSingletonMainInput(transform), new GroupAndWriteView<>(view)); }  @Override PTransformReplacement<PCollection<ElemT>, PCollection<ElemT>> getReplacementTransform(
      AppliedPTransform<
              PCollection<ElemT>,
              PCollection<ElemT>,
              PTransform<PCollection<ElemT>, PCollection<ElemT>>>
          transform); @Override Map<PValue, ReplacementOutput> mapOutputs(
      Map<TupleTag<?>, PValue> outputs, PCollection<ElemT> newOutput); static final String DIRECT_WRITE_VIEW_URN; }
@Test public void replacementGetViewReturnsOriginal() { final PCollection<Integer> ints = p.apply("CreateContents", Create.of(1, 2, 3)); final PCollectionView<List<Integer>> view = ints.apply(View.asList()); PTransformReplacement<PCollection<Integer>, PCollection<Integer>> replacement = factory.getReplacementTransform( AppliedPTransform.of( "foo", ints.expand(), view.expand(), CreatePCollectionView.of(view), p)); ints.apply(replacement.getTransform()); final AtomicBoolean writeViewVisited = new AtomicBoolean(); p.traverseTopologically( new PipelineVisitor.Defaults() { @Override public void visitPrimitiveTransform(Node node) { if (node.getTransform() instanceof WriteView) { assertThat( "There should only be one WriteView primitive in the graph", writeViewVisited.getAndSet(true), is(false)); PCollectionView<?> replacementView = ((WriteView) node.getTransform()).getView(); assertThat( replacementView.getTagInternal(), equalTo((TupleTag) view.getTagInternal())); assertThat(replacementView.getViewFn(), equalTo(view.getViewFn())); assertThat(replacementView.getWindowMappingFn(), equalTo(view.getWindowMappingFn())); assertThat(node.getInputs().entrySet(), hasSize(1)); } } }); assertThat(writeViewVisited.get(), is(true)); }