View { public static <T> AsList<T> asList() { return new AsList<>(); } 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 testListSideInput() { final PCollectionView<List<Integer>> view = pipeline.apply("CreateSideInput", Create.of(11, 13, 17, 23)).apply(View.asList()); PCollection<Integer> output = pipeline .apply("CreateMainInput", Create.of(29, 31)) .apply( "OutputSideInputs", ParDo.of( new DoFn<Integer, Integer>() { @ProcessElement public void processElement(ProcessContext c) { checkArgument(c.sideInput(view).size() == 4); checkArgument( c.sideInput(view).get(0).equals(c.sideInput(view).get(0))); 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 testWindowedListSideInput() { final PCollectionView<List<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.asList()); 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) { checkArgument(c.sideInput(view).size() == 4); checkArgument( c.sideInput(view).get(0).equals(c.sideInput(view).get(0))); 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 testEmptyListSideInput() throws Exception { final PCollectionView<List<Integer>> view = pipeline.apply("CreateEmptyView", Create.empty(VarIntCoder.of())).apply(View.asList()); PCollection<Integer> results = pipeline .apply("Create1", Create.of(1)) .apply( "OutputSideInputs", ParDo.of( new DoFn<Integer, Integer>() { @ProcessElement public void processElement(ProcessContext c) { assertTrue(c.sideInput(view).isEmpty()); assertFalse(c.sideInput(view).iterator().hasNext()); c.output(1); } }) .withSideInputs(view)); PAssert.that(results).containsInAnyOrder(1); pipeline.run(); }
@Test @Category(ValidatesRunner.class) public void testListSideInputIsImmutable() { final PCollectionView<List<Integer>> view = pipeline.apply("CreateSideInput", Create.of(11)).apply(View.asList()); PCollection<Integer> output = pipeline .apply("CreateMainInput", Create.of(29)) .apply( "OutputSideInputs", ParDo.of( new DoFn<Integer, Integer>() { @ProcessElement public void processElement(ProcessContext c) { try { c.sideInput(view).clear(); fail("Expected UnsupportedOperationException on clear()"); } catch (UnsupportedOperationException expected) { } try { c.sideInput(view).add(4); fail("Expected UnsupportedOperationException on add()"); } catch (UnsupportedOperationException expected) { } try { c.sideInput(view).addAll(new ArrayList<>()); fail("Expected UnsupportedOperationException on addAll()"); } catch (UnsupportedOperationException expected) { } try { c.sideInput(view).remove(0); fail("Expected UnsupportedOperationException on remove()"); } catch (UnsupportedOperationException expected) { } for (Integer i : c.sideInput(view)) { c.output(i); } } }) .withSideInputs(view)); PAssert.that(output).containsInAnyOrder(11); pipeline.run(); }
@Test public void testViewUnboundedAsListDirect() { testViewUnbounded(pipeline, View.asList()); }
@Test public void testViewNonmergingAsListDirect() { testViewNonmerging(pipeline, View.asList()); }