View { public static <T> AsIterable<T> asIterable() { return new AsIterable<>(); } 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 testIterableSideInput() { final PCollectionView<Iterable<Integer>> view = pipeline.apply("CreateSideInput", Create.of(11, 13, 17, 23)).apply(View.asIterable()); PCollection<Integer> output = pipeline .apply("CreateMainInput", Create.of(29, 31)) .apply( "OutputSideInputs", ParDo.of( new DoFn<Integer, Integer>() { @ProcessElement public void processElement(ProcessContext c) { for (Integer i : c.sideInput(view)) { c.output(i); } } }) .withSideInputs(view)); PAssert.that(output).containsInAnyOrder(11, 13, 17, 23, 11, 13, 17, 23); pipeline.run(); }
@Test @Category(ValidatesRunner.class) public void testWindowedIterableSideInput() { final PCollectionView<Iterable<Integer>> view = pipeline .apply( "CreateSideInput", Create.timestamped( TimestampedValue.of(11, new Instant(1)), TimestampedValue.of(13, new Instant(1)), TimestampedValue.of(17, new Instant(1)), TimestampedValue.of(23, new Instant(1)), TimestampedValue.of(31, new Instant(11)), TimestampedValue.of(33, new Instant(11)), TimestampedValue.of(37, new Instant(11)), TimestampedValue.of(43, new Instant(11)))) .apply("SideWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) .apply(View.asIterable()); PCollection<Integer> output = pipeline .apply( "CreateMainInput", Create.timestamped( TimestampedValue.of(29, new Instant(1)), TimestampedValue.of(35, new Instant(11)))) .apply("MainWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) .apply( "OutputSideInputs", ParDo.of( new DoFn<Integer, Integer>() { @ProcessElement public void processElement(ProcessContext c) { for (Integer i : c.sideInput(view)) { c.output(i); } } }) .withSideInputs(view)); PAssert.that(output).containsInAnyOrder(11, 13, 17, 23, 31, 33, 37, 43); pipeline.run(); }
@Test @Category(ValidatesRunner.class) public void testEmptyIterableSideInput() throws Exception { final PCollectionView<Iterable<Integer>> view = pipeline.apply("CreateEmptyView", Create.empty(VarIntCoder.of())).apply(View.asIterable()); PCollection<Integer> results = pipeline .apply("Create1", Create.of(1)) .apply( "OutputSideInputs", ParDo.of( new DoFn<Integer, Integer>() { @ProcessElement public void processElement(ProcessContext c) { assertFalse(c.sideInput(view).iterator().hasNext()); c.output(1); } }) .withSideInputs(view)); PAssert.that(results).containsInAnyOrder(1); pipeline.run(); }
@Test @Category(ValidatesRunner.class) public void testIterableSideInputIsImmutable() { final PCollectionView<Iterable<Integer>> view = pipeline.apply("CreateSideInput", Create.of(11)).apply(View.asIterable()); PCollection<Integer> output = pipeline .apply("CreateMainInput", Create.of(29)) .apply( "OutputSideInputs", ParDo.of( new DoFn<Integer, Integer>() { @ProcessElement public void processElement(ProcessContext c) { Iterator<Integer> iterator = c.sideInput(view).iterator(); while (iterator.hasNext()) { try { iterator.remove(); fail("Expected UnsupportedOperationException on remove()"); } catch (UnsupportedOperationException expected) { } c.output(iterator.next()); } } }) .withSideInputs(view)); PAssert.that(output).containsInAnyOrder(11); pipeline.run(); }
@Test @Category(ValidatesRunner.class) public void testSideInputWithNestedIterables() { final PCollectionView<Iterable<Integer>> view1 = pipeline .apply("CreateVoid1", Create.of((Void) null).withCoder(VoidCoder.of())) .apply( "OutputOneInteger", ParDo.of( new DoFn<Void, Integer>() { @ProcessElement public void processElement(ProcessContext c) { c.output(17); } })) .apply("View1", View.asIterable()); final PCollectionView<Iterable<Iterable<Integer>>> view2 = pipeline .apply("CreateVoid2", Create.of((Void) null).withCoder(VoidCoder.of())) .apply( "OutputSideInput", ParDo.of( new DoFn<Void, Iterable<Integer>>() { @ProcessElement public void processElement(ProcessContext c) { c.output(c.sideInput(view1)); } }) .withSideInputs(view1)) .apply("View2", View.asIterable()); PCollection<Integer> output = pipeline .apply("CreateVoid3", Create.of((Void) null).withCoder(VoidCoder.of())) .apply( "ReadIterableSideInput", ParDo.of( new DoFn<Void, Integer>() { @ProcessElement public void processElement(ProcessContext c) { for (Iterable<Integer> input : c.sideInput(view2)) { for (Integer i : input) { c.output(i); } } } }) .withSideInputs(view2)); PAssert.that(output).containsInAnyOrder(17); pipeline.run(); }
@Test public void testViewUnboundedAsIterableDirect() { testViewUnbounded(pipeline, View.asIterable()); }
@Test public void testViewNonmergingAsIterableDirect() { testViewNonmerging(pipeline, View.asIterable()); }