CombineFns { public static ComposeCombineFnBuilder compose() { return new ComposeCombineFnBuilder(); }  static ComposeCombineFnBuilder compose();  }
@Test public void testDuplicatedTags() { expectedException.expect(IllegalArgumentException.class); expectedException.expectMessage("it is already present in the composition"); TupleTag<Integer> tag = new TupleTag<>(); CombineFns.compose() .with(new GetIntegerFunction(), Max.ofIntegers(), tag) .with(new GetIntegerFunction(), Min.ofIntegers(), tag); }
@Test public void testDuplicatedTagsKeyed() { expectedException.expect(IllegalArgumentException.class); expectedException.expectMessage("it is already present in the composition"); TupleTag<Integer> tag = new TupleTag<>(); CombineFns.compose() .with(new GetIntegerFunction(), Max.ofIntegers(), tag) .with(new GetIntegerFunction(), Min.ofIntegers(), tag); }
@Test public void testDuplicatedTagsWithContext() { expectedException.expect(IllegalArgumentException.class); expectedException.expectMessage("it is already present in the composition"); TupleTag<UserString> tag = new TupleTag<>(); CombineFns.compose() .with(new GetUserStringFunction(), new ConcatStringWithContext(null ), tag) .with(new GetUserStringFunction(), new ConcatStringWithContext(null ), tag); }
@Test @Category({ValidatesRunner.class, UsesSideInputs.class}) public void testComposedCombine() { p.getCoderRegistry().registerCoderForClass(UserString.class, UserStringCoder.of()); PCollection<KV<String, KV<Integer, UserString>>> perKeyInput = p.apply( Create.timestamped( Arrays.asList( KV.of("a", KV.of(1, UserString.of("1"))), KV.of("a", KV.of(1, UserString.of("1"))), KV.of("a", KV.of(4, UserString.of("4"))), KV.of("b", KV.of(1, UserString.of("1"))), KV.of("b", KV.of(13, UserString.of("13")))), Arrays.asList(0L, 4L, 7L, 10L, 16L)) .withCoder( KvCoder.of( StringUtf8Coder.of(), KvCoder.of(BigEndianIntegerCoder.of(), UserStringCoder.of())))); TupleTag<Integer> maxIntTag = new TupleTag<>(); TupleTag<UserString> concatStringTag = new TupleTag<>(); PCollection<KV<String, KV<Integer, String>>> combineGlobally = perKeyInput .apply(Values.create()) .apply( Combine.globally( CombineFns.compose() .with(new GetIntegerFunction(), Max.ofIntegers(), maxIntTag) .with(new GetUserStringFunction(), new ConcatString(), concatStringTag))) .apply(WithKeys.of("global")) .apply( "ExtractGloballyResult", ParDo.of(new ExtractResultDoFn(maxIntTag, concatStringTag))); PCollection<KV<String, KV<Integer, String>>> combinePerKey = perKeyInput .apply( Combine.perKey( CombineFns.compose() .with(new GetIntegerFunction(), Max.ofIntegers(), maxIntTag) .with(new GetUserStringFunction(), new ConcatString(), concatStringTag))) .apply( "ExtractPerKeyResult", ParDo.of(new ExtractResultDoFn(maxIntTag, concatStringTag))); PAssert.that(combineGlobally).containsInAnyOrder(KV.of("global", KV.of(13, "111134"))); PAssert.that(combinePerKey) .containsInAnyOrder(KV.of("a", KV.of(4, "114")), KV.of("b", KV.of(13, "113"))); p.run(); }
@Test @Category({ValidatesRunner.class, UsesSideInputs.class}) public void testComposedCombineWithContext() { p.getCoderRegistry().registerCoderForClass(UserString.class, UserStringCoder.of()); PCollectionView<String> view = p.apply(Create.of("I")).apply(View.asSingleton()); PCollection<KV<String, KV<Integer, UserString>>> perKeyInput = p.apply( Create.timestamped( Arrays.asList( KV.of("a", KV.of(1, UserString.of("1"))), KV.of("a", KV.of(1, UserString.of("1"))), KV.of("a", KV.of(4, UserString.of("4"))), KV.of("b", KV.of(1, UserString.of("1"))), KV.of("b", KV.of(13, UserString.of("13")))), Arrays.asList(0L, 4L, 7L, 10L, 16L)) .withCoder( KvCoder.of( StringUtf8Coder.of(), KvCoder.of(BigEndianIntegerCoder.of(), UserStringCoder.of())))); TupleTag<Integer> maxIntTag = new TupleTag<>(); TupleTag<UserString> concatStringTag = new TupleTag<>(); PCollection<KV<String, KV<Integer, String>>> combineGlobally = perKeyInput .apply(Values.create()) .apply( Combine.globally( CombineFns.compose() .with(new GetIntegerFunction(), Max.ofIntegers(), maxIntTag) .with( new GetUserStringFunction(), new ConcatStringWithContext(view), concatStringTag)) .withoutDefaults() .withSideInputs(ImmutableList.of(view))) .apply(WithKeys.of("global")) .apply( "ExtractGloballyResult", ParDo.of(new ExtractResultDoFn(maxIntTag, concatStringTag))); PCollection<KV<String, KV<Integer, String>>> combinePerKey = perKeyInput .apply( Combine.<String, KV<Integer, UserString>, CoCombineResult>perKey( CombineFns.compose() .with(new GetIntegerFunction(), Max.ofIntegers(), maxIntTag) .with( new GetUserStringFunction(), new ConcatStringWithContext(view), concatStringTag)) .withSideInputs(ImmutableList.of(view))) .apply( "ExtractPerKeyResult", ParDo.of(new ExtractResultDoFn(maxIntTag, concatStringTag))); PAssert.that(combineGlobally).containsInAnyOrder(KV.of("global", KV.of(13, "111134I"))); PAssert.that(combinePerKey) .containsInAnyOrder(KV.of("a", KV.of(4, "114I")), KV.of("b", KV.of(13, "113I"))); p.run(); }
@Test @Category({ValidatesRunner.class}) public void testComposedCombineNullValues() { p.getCoderRegistry() .registerCoderForClass(UserString.class, NullableCoder.of(UserStringCoder.of())); p.getCoderRegistry() .registerCoderForClass(String.class, NullableCoder.of(StringUtf8Coder.of())); PCollection<KV<String, KV<Integer, UserString>>> perKeyInput = p.apply( Create.timestamped( Arrays.asList( KV.of("a", KV.of(1, UserString.of("1"))), KV.of("a", KV.of(1, UserString.of("1"))), KV.of("a", KV.of(4, UserString.of("4"))), KV.of("b", KV.of(1, UserString.of("1"))), KV.of("b", KV.of(13, UserString.of("13")))), Arrays.asList(0L, 4L, 7L, 10L, 16L)) .withCoder( KvCoder.of( NullableCoder.of(StringUtf8Coder.of()), KvCoder.of( BigEndianIntegerCoder.of(), NullableCoder.of(UserStringCoder.of()))))); TupleTag<Integer> maxIntTag = new TupleTag<>(); TupleTag<UserString> concatStringTag = new TupleTag<>(); PCollection<KV<String, KV<Integer, String>>> combinePerKey = perKeyInput .apply( Combine.perKey( CombineFns.compose() .with(new GetIntegerFunction(), Max.ofIntegers(), maxIntTag) .with( new GetUserStringFunction(), new OutputNullString(), concatStringTag))) .apply( "ExtractPerKeyResult", ParDo.of(new ExtractResultDoFn(maxIntTag, concatStringTag))); PAssert.that(combinePerKey) .containsInAnyOrder( KV.of("a", KV.of(4, (String) null)), KV.of("b", KV.of(13, (String) null))); p.run(); }
@Test public void testComposedCombineDisplayData() { SimpleFunction<String, String> extractFn = new SimpleFunction<String, String>() { @Override public String apply(String input) { return input; } }; DisplayDataCombineFn combineFn1 = new DisplayDataCombineFn("value1"); DisplayDataCombineFn combineFn2 = new DisplayDataCombineFn("value2"); CombineFns.ComposedCombineFn<String> composedCombine = CombineFns.compose() .with(extractFn, combineFn1, new TupleTag<>()) .with(extractFn, combineFn2, new TupleTag<>()); DisplayData displayData = DisplayData.from(composedCombine); assertThat(displayData, hasDisplayItem("combineFn1", combineFn1.getClass())); assertThat(displayData, hasDisplayItem("combineFn2", combineFn2.getClass())); assertThat(displayData, includesDisplayDataFor("combineFn1", combineFn1)); assertThat(displayData, includesDisplayDataFor("combineFn2", combineFn2)); }
@Test @Category({ValidatesRunner.class, UsesSideInputs.class, DataflowPortabilityApiUnsupported.class}) public void testComposedCombine() { p.getCoderRegistry().registerCoderForClass(UserString.class, UserStringCoder.of()); PCollection<KV<String, KV<Integer, UserString>>> perKeyInput = p.apply( Create.timestamped( Arrays.asList( KV.of("a", KV.of(1, UserString.of("1"))), KV.of("a", KV.of(1, UserString.of("1"))), KV.of("a", KV.of(4, UserString.of("4"))), KV.of("b", KV.of(1, UserString.of("1"))), KV.of("b", KV.of(13, UserString.of("13")))), Arrays.asList(0L, 4L, 7L, 10L, 16L)) .withCoder( KvCoder.of( StringUtf8Coder.of(), KvCoder.of(BigEndianIntegerCoder.of(), UserStringCoder.of())))); TupleTag<Integer> maxIntTag = new TupleTag<>(); TupleTag<UserString> concatStringTag = new TupleTag<>(); PCollection<KV<String, KV<Integer, String>>> combineGlobally = perKeyInput .apply(Values.create()) .apply( Combine.globally( CombineFns.compose() .with(new GetIntegerFunction(), Max.ofIntegers(), maxIntTag) .with(new GetUserStringFunction(), new ConcatString(), concatStringTag))) .apply(WithKeys.of("global")) .apply( "ExtractGloballyResult", ParDo.of(new ExtractResultDoFn(maxIntTag, concatStringTag))); PCollection<KV<String, KV<Integer, String>>> combinePerKey = perKeyInput .apply( Combine.perKey( CombineFns.compose() .with(new GetIntegerFunction(), Max.ofIntegers(), maxIntTag) .with(new GetUserStringFunction(), new ConcatString(), concatStringTag))) .apply( "ExtractPerKeyResult", ParDo.of(new ExtractResultDoFn(maxIntTag, concatStringTag))); PAssert.that(combineGlobally).containsInAnyOrder(KV.of("global", KV.of(13, "111134"))); PAssert.that(combinePerKey) .containsInAnyOrder(KV.of("a", KV.of(4, "114")), KV.of("b", KV.of(13, "113"))); p.run(); }
@Test @Category({ValidatesRunner.class, UsesSideInputs.class, DataflowPortabilityApiUnsupported.class}) public void testComposedCombineWithContext() { p.getCoderRegistry().registerCoderForClass(UserString.class, UserStringCoder.of()); PCollectionView<String> view = p.apply(Create.of("I")).apply(View.asSingleton()); PCollection<KV<String, KV<Integer, UserString>>> perKeyInput = p.apply( Create.timestamped( Arrays.asList( KV.of("a", KV.of(1, UserString.of("1"))), KV.of("a", KV.of(1, UserString.of("1"))), KV.of("a", KV.of(4, UserString.of("4"))), KV.of("b", KV.of(1, UserString.of("1"))), KV.of("b", KV.of(13, UserString.of("13")))), Arrays.asList(0L, 4L, 7L, 10L, 16L)) .withCoder( KvCoder.of( StringUtf8Coder.of(), KvCoder.of(BigEndianIntegerCoder.of(), UserStringCoder.of())))); TupleTag<Integer> maxIntTag = new TupleTag<>(); TupleTag<UserString> concatStringTag = new TupleTag<>(); PCollection<KV<String, KV<Integer, String>>> combineGlobally = perKeyInput .apply(Values.create()) .apply( Combine.globally( CombineFns.compose() .with(new GetIntegerFunction(), Max.ofIntegers(), maxIntTag) .with( new GetUserStringFunction(), new ConcatStringWithContext(view), concatStringTag)) .withoutDefaults() .withSideInputs(ImmutableList.of(view))) .apply(WithKeys.of("global")) .apply( "ExtractGloballyResult", ParDo.of(new ExtractResultDoFn(maxIntTag, concatStringTag))); PCollection<KV<String, KV<Integer, String>>> combinePerKey = perKeyInput .apply( Combine.<String, KV<Integer, UserString>, CoCombineResult>perKey( CombineFns.compose() .with(new GetIntegerFunction(), Max.ofIntegers(), maxIntTag) .with( new GetUserStringFunction(), new ConcatStringWithContext(view), concatStringTag)) .withSideInputs(ImmutableList.of(view))) .apply( "ExtractPerKeyResult", ParDo.of(new ExtractResultDoFn(maxIntTag, concatStringTag))); PAssert.that(combineGlobally).containsInAnyOrder(KV.of("global", KV.of(13, "111134I"))); PAssert.that(combinePerKey) .containsInAnyOrder(KV.of("a", KV.of(4, "114I")), KV.of("b", KV.of(13, "113I"))); p.run(); }
@Test @Category({ValidatesRunner.class, DataflowPortabilityApiUnsupported.class}) public void testComposedCombineNullValues() { p.getCoderRegistry() .registerCoderForClass(UserString.class, NullableCoder.of(UserStringCoder.of())); p.getCoderRegistry() .registerCoderForClass(String.class, NullableCoder.of(StringUtf8Coder.of())); PCollection<KV<String, KV<Integer, UserString>>> perKeyInput = p.apply( Create.timestamped( Arrays.asList( KV.of("a", KV.of(1, UserString.of("1"))), KV.of("a", KV.of(1, UserString.of("1"))), KV.of("a", KV.of(4, UserString.of("4"))), KV.of("b", KV.of(1, UserString.of("1"))), KV.of("b", KV.of(13, UserString.of("13")))), Arrays.asList(0L, 4L, 7L, 10L, 16L)) .withCoder( KvCoder.of( NullableCoder.of(StringUtf8Coder.of()), KvCoder.of( BigEndianIntegerCoder.of(), NullableCoder.of(UserStringCoder.of()))))); TupleTag<Integer> maxIntTag = new TupleTag<>(); TupleTag<UserString> concatStringTag = new TupleTag<>(); PCollection<KV<String, KV<Integer, String>>> combinePerKey = perKeyInput .apply( Combine.perKey( CombineFns.compose() .with(new GetIntegerFunction(), Max.ofIntegers(), maxIntTag) .with( new GetUserStringFunction(), new OutputNullString(), concatStringTag))) .apply( "ExtractPerKeyResult", ParDo.of(new ExtractResultDoFn(maxIntTag, concatStringTag))); PAssert.that(combinePerKey) .containsInAnyOrder( KV.of("a", KV.of(4, (String) null)), KV.of("b", KV.of(13, (String) null))); p.run(); }