[BEAM-6719] Allow multiple Joins in the same pipeline #7813

DanielMe · 2019-02-11T21:34:19Z

This change wraps the joins from the joinlibrary extension in individual PTransforms. I also provide overloaded methods which allow to name the corresponding nodes in the graph.

The background of this change is that currently it is not possible to have multiple joins in the same pipeline without wrapping them in individual PTransforms as this would generate name clashes.

Consider the following test case:

 @Test
  public void testMultipleJoinsInSamePipeline() {
    leftListOfKv.add(KV.of("Key2", 4L));
    PCollection<KV<String, Long>> leftCollection = p.apply("CreateLeft", Create.of(leftListOfKv));

    rightListOfKv.add(KV.of("Key2", "bar"));
    PCollection<KV<String, String>> rightCollection =
            p.apply("CreateRight", Create.of(rightListOfKv));

    expectedResult.add(KV.of("Key2", KV.of(4L, "bar")));

    PCollection<KV<String, KV<Long, String>>> output1 =
            Join.innerJoin(leftCollection, rightCollection);
    PCollection<KV<String, KV<Long, String>>> output2 =
            Join.innerJoin(leftCollection, rightCollection);
    PAssert.that(output1).containsInAnyOrder(expectedResult);
    PAssert.that(output2).containsInAnyOrder(expectedResult);

    p.run();
  }

With the change contained in this PR, the same code would still fail but there is now an overloaded call to Join.innerJoin so that the corresponding nodes in the execution graph receive different names (see test case below).

The change is backwards compatible. Two other side benefits are:

The naming of the transformation is slightly more intuitive in case you want to debug / look at the execution graph it's now easier to see which steps correspond to the join.
It also allows using the PTransform directly writing something like myCollection1.apply(InnerJoin.with(myCollection2)) which some people might find more intuitive.
Choose reviewer(s) and mention them in a comment (R: @username).
Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
If this contribution is large, please file an Apache Individual Contributor License Agreement.