View { public static <T> AsSingleton<T> asSingleton() { return new AsSingleton<>(); } private  View(); static AsSingleton<T> asSingleton(); static AsList<T> asList(); static AsIterable<T> asIterable(); static AsMap<K, V> asMap(); static AsMultimap<K, V> asMultimap();  }
@Test @Category(ValidatesRunner.class) public void testSingletonSideInput() { final PCollectionView<Integer> view = pipeline.apply("Create47", Create.of(47)).apply(View.asSingleton()); PCollection<Integer> output = pipeline .apply("Create123", Create.of(1, 2, 3)) .apply( "OutputSideInputs", ParDo.of( new DoFn<Integer, Integer>() { @ProcessElement public void processElement(ProcessContext c) { c.output(c.sideInput(view)); } }) .withSideInputs(view)); PAssert.that(output).containsInAnyOrder(47, 47, 47); pipeline.run(); }
@Test @Category(ValidatesRunner.class) public void testWindowedSingletonSideInput() { final PCollectionView<Integer> view = pipeline .apply( "Create47", Create.timestamped( TimestampedValue.of(47, new Instant(1)), TimestampedValue.of(48, new Instant(11)))) .apply("SideWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) .apply(View.asSingleton()); PCollection<Integer> output = pipeline .apply( "Create123", Create.timestamped( TimestampedValue.of(1, new Instant(4)), TimestampedValue.of(2, new Instant(8)), TimestampedValue.of(3, new Instant(12)))) .apply("MainWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) .apply( "OutputSideInputs", ParDo.of( new DoFn<Integer, Integer>() { @ProcessElement public void processElement(ProcessContext c) { c.output(c.sideInput(view)); } }) .withSideInputs(view)); PAssert.that(output).containsInAnyOrder(47, 47, 48); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void testEmptySingletonSideInput() throws Exception { final PCollectionView<Integer> view = pipeline .apply("CreateEmptyIntegers", Create.empty(VarIntCoder.of())) .apply(View.asSingleton()); pipeline .apply("Create123", Create.of(1, 2, 3)) .apply( "OutputSideInputs", ParDo.of( new DoFn<Integer, Integer>() { @ProcessElement public void processElement(ProcessContext c) { c.output(c.sideInput(view)); } }) .withSideInputs(view)); thrown.expect(PipelineExecutionException.class); thrown.expectCause(isA(NoSuchElementException.class)); thrown.expectMessage("Empty"); thrown.expectMessage("PCollection"); thrown.expectMessage("singleton"); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void testNonSingletonSideInput() throws Exception { PCollection<Integer> oneTwoThree = pipeline.apply(Create.of(1, 2, 3)); final PCollectionView<Integer> view = oneTwoThree.apply(View.asSingleton()); oneTwoThree.apply( "OutputSideInputs", ParDo.of( new DoFn<Integer, Integer>() { @ProcessElement public void processElement(ProcessContext c) { c.output(c.sideInput(view)); } }) .withSideInputs(view)); thrown.expect(PipelineExecutionException.class); thrown.expectCause(isA(IllegalArgumentException.class)); thrown.expectMessage("PCollection"); thrown.expectMessage("more than one"); thrown.expectMessage("singleton"); pipeline.run(); }
@Test @Category(ValidatesRunner.class) public void testWindowedSideInputFixedToFixed() { final PCollectionView<Integer> view = pipeline .apply( "CreateSideInput", Create.timestamped( TimestampedValue.of(1, new Instant(1)), TimestampedValue.of(2, new Instant(11)), TimestampedValue.of(3, new Instant(13)))) .apply("WindowSideInput", Window.into(FixedWindows.of(Duration.millis(10)))) .apply(Sum.integersGlobally().withoutDefaults()) .apply(View.asSingleton()); PCollection<String> output = pipeline .apply( "CreateMainInput", Create.timestamped( TimestampedValue.of("A", new Instant(4)), TimestampedValue.of("B", new Instant(15)), TimestampedValue.of("C", new Instant(7)))) .apply("WindowMainInput", Window.into(FixedWindows.of(Duration.millis(10)))) .apply( "OutputMainAndSideInputs", ParDo.of( new DoFn<String, String>() { @ProcessElement public void processElement(ProcessContext c) { c.output(c.element() + c.sideInput(view)); } }) .withSideInputs(view)); PAssert.that(output).containsInAnyOrder("A1", "B5", "C1"); pipeline.run(); }
@Test @Category(ValidatesRunner.class) public void testWindowedSideInputFixedToGlobal() { final PCollectionView<Integer> view = pipeline .apply( "CreateSideInput", Create.timestamped( TimestampedValue.of(1, new Instant(1)), TimestampedValue.of(2, new Instant(11)), TimestampedValue.of(3, new Instant(13)))) .apply("WindowSideInput", Window.into(new GlobalWindows())) .apply(Sum.integersGlobally()) .apply(View.asSingleton()); PCollection<String> output = pipeline .apply( "CreateMainInput", Create.timestamped( TimestampedValue.of("A", new Instant(4)), TimestampedValue.of("B", new Instant(15)), TimestampedValue.of("C", new Instant(7)))) .apply("WindowMainInput", Window.into(FixedWindows.of(Duration.millis(10)))) .apply( "OutputMainAndSideInputs", ParDo.of( new DoFn<String, String>() { @ProcessElement public void processElement(ProcessContext c) { c.output(c.element() + c.sideInput(view)); } }) .withSideInputs(view)); PAssert.that(output).containsInAnyOrder("A6", "B6", "C6"); pipeline.run(); }
@Test public void testViewUnboundedAsSingletonDirect() { testViewUnbounded(pipeline, View.asSingleton()); }
@Test public void testViewNonmergingAsSingletonDirect() { testViewNonmerging(pipeline, View.asSingleton()); }