@Test public void testGetMaterializeSourceTarget_AlreadyMaterialized() { PCollectionImpl<String> materializedPcollection = mock(PCollectionImpl.class); ReadableSourceTarget<String> readableSourceTarget = mock(ReadableSourceTarget.class); when(materializedPcollection.getMaterializedAt()).thenReturn(readableSourceTarget); assertEquals(readableSourceTarget, pipeline.getMaterializeSourceTarget(materializedPcollection)); }
public <T> ReadableSourceTarget<T> getMaterializeSourceTarget(PCollection<T> pcollection) { PCollectionImpl<T> impl = toPcollectionImpl(pcollection); SourceTarget<T> matTarget = impl.getMaterializedAt(); if (matTarget != null && matTarget instanceof ReadableSourceTarget) { return (ReadableSourceTarget<T>) matTarget; } ReadableSourceTarget<T> srcTarget = null; if (outputTargets.containsKey(pcollection)) { for (Target target : outputTargets.get(impl)) { if (target instanceof ReadableSourceTarget) { srcTarget = (ReadableSourceTarget<T>) target; break; } } } if (srcTarget == null) { SourceTarget<T> st = createIntermediateOutput(pcollection.getPType()); if (!(st instanceof ReadableSourceTarget)) { throw new IllegalArgumentException("The PType for the given PCollection is not readable" + " and cannot be materialized"); } else { srcTarget = (ReadableSourceTarget<T>) st; addOutput(impl, srcTarget); } } return srcTarget; }
MRPipeline implements Pipeline { public <T> ReadableSourceTarget<T> getMaterializeSourceTarget(PCollection<T> pcollection) { PCollectionImpl<T> impl = toPcollectionImpl(pcollection); SourceTarget<T> matTarget = impl.getMaterializedAt(); if (matTarget != null && matTarget instanceof ReadableSourceTarget) { return (ReadableSourceTarget<T>) matTarget; } ReadableSourceTarget<T> srcTarget = null; if (outputTargets.containsKey(pcollection)) { for (Target target : outputTargets.get(impl)) { if (target instanceof ReadableSourceTarget) { srcTarget = (ReadableSourceTarget<T>) target; break; } } } if (srcTarget == null) { SourceTarget<T> st = createIntermediateOutput(pcollection.getPType()); if (!(st instanceof ReadableSourceTarget)) { throw new IllegalArgumentException("The PType for the given PCollection is not readable" + " and cannot be materialized"); } else { srcTarget = (ReadableSourceTarget<T>) st; addOutput(impl, srcTarget); } } return srcTarget; } }
MRPipeline implements Pipeline { public <T> ReadableSourceTarget<T> getMaterializeSourceTarget(PCollection<T> pcollection) { PCollectionImpl<T> impl = toPcollectionImpl(pcollection); SourceTarget<T> matTarget = impl.getMaterializedAt(); if (matTarget != null && matTarget instanceof ReadableSourceTarget) { return (ReadableSourceTarget<T>) matTarget; } ReadableSourceTarget<T> srcTarget = null; if (outputTargets.containsKey(pcollection)) { for (Target target : outputTargets.get(impl)) { if (target instanceof ReadableSourceTarget) { srcTarget = (ReadableSourceTarget<T>) target; break; } } } if (srcTarget == null) { SourceTarget<T> st = createIntermediateOutput(pcollection.getPType()); if (!(st instanceof ReadableSourceTarget)) { throw new IllegalArgumentException("The PType for the given PCollection is not readable" + " and cannot be materialized"); } else { srcTarget = (ReadableSourceTarget<T>) st; addOutput(impl, srcTarget); } } return srcTarget; } MRPipeline(Class<?> jarClass);  MRPipeline(Class<?> jarClass, String name);  MRPipeline(Class<?> jarClass, Configuration conf);  MRPipeline(Class<?> jarClass, String name, Configuration conf); }
MRPipeline implements Pipeline { public <T> ReadableSourceTarget<T> getMaterializeSourceTarget(PCollection<T> pcollection) { PCollectionImpl<T> impl = toPcollectionImpl(pcollection); SourceTarget<T> matTarget = impl.getMaterializedAt(); if (matTarget != null && matTarget instanceof ReadableSourceTarget) { return (ReadableSourceTarget<T>) matTarget; } ReadableSourceTarget<T> srcTarget = null; if (outputTargets.containsKey(pcollection)) { for (Target target : outputTargets.get(impl)) { if (target instanceof ReadableSourceTarget) { srcTarget = (ReadableSourceTarget<T>) target; break; } } } if (srcTarget == null) { SourceTarget<T> st = createIntermediateOutput(pcollection.getPType()); if (!(st instanceof ReadableSourceTarget)) { throw new IllegalArgumentException("The PType for the given PCollection is not readable" + " and cannot be materialized"); } else { srcTarget = (ReadableSourceTarget<T>) st; addOutput(impl, srcTarget); } } return srcTarget; } MRPipeline(Class<?> jarClass);  MRPipeline(Class<?> jarClass, String name);  MRPipeline(Class<?> jarClass, Configuration conf);  MRPipeline(Class<?> jarClass, String name, Configuration conf); @Override Configuration getConfiguration(); @Override void setConfiguration(Configuration conf); @Override PipelineResult run(); @Override PipelineResult done(); PCollection<S> read(Source<S> source); PTable<K, V> read(TableSource<K, V> source); PCollection<String> readTextFile(String pathName); @SuppressWarnings("unchecked") void write(PCollection<?> pcollection, Target target); @Override Iterable<T> materialize(PCollection<T> pcollection); ReadableSourceTarget<T> getMaterializeSourceTarget(PCollection<T> pcollection); SourceTarget<T> createIntermediateOutput(PType<T> ptype); Path createTempPath(); @Override void writeTextFile(PCollection<T> pcollection, String pathName); int getNextAnonymousStageId(); @Override void enableDebug(); @Override String getName(); }
MRPipeline implements Pipeline { public <T> ReadableSourceTarget<T> getMaterializeSourceTarget(PCollection<T> pcollection) { PCollectionImpl<T> impl = toPcollectionImpl(pcollection); SourceTarget<T> matTarget = impl.getMaterializedAt(); if (matTarget != null && matTarget instanceof ReadableSourceTarget) { return (ReadableSourceTarget<T>) matTarget; } ReadableSourceTarget<T> srcTarget = null; if (outputTargets.containsKey(pcollection)) { for (Target target : outputTargets.get(impl)) { if (target instanceof ReadableSourceTarget) { srcTarget = (ReadableSourceTarget<T>) target; break; } } } if (srcTarget == null) { SourceTarget<T> st = createIntermediateOutput(pcollection.getPType()); if (!(st instanceof ReadableSourceTarget)) { throw new IllegalArgumentException("The PType for the given PCollection is not readable" + " and cannot be materialized"); } else { srcTarget = (ReadableSourceTarget<T>) st; addOutput(impl, srcTarget); } } return srcTarget; } MRPipeline(Class<?> jarClass);  MRPipeline(Class<?> jarClass, String name);  MRPipeline(Class<?> jarClass, Configuration conf);  MRPipeline(Class<?> jarClass, String name, Configuration conf); @Override Configuration getConfiguration(); @Override void setConfiguration(Configuration conf); @Override PipelineResult run(); @Override PipelineResult done(); PCollection<S> read(Source<S> source); PTable<K, V> read(TableSource<K, V> source); PCollection<String> readTextFile(String pathName); @SuppressWarnings("unchecked") void write(PCollection<?> pcollection, Target target); @Override Iterable<T> materialize(PCollection<T> pcollection); ReadableSourceTarget<T> getMaterializeSourceTarget(PCollection<T> pcollection); SourceTarget<T> createIntermediateOutput(PType<T> ptype); Path createTempPath(); @Override void writeTextFile(PCollection<T> pcollection, String pathName); int getNextAnonymousStageId(); @Override void enableDebug(); @Override String getName();  }