public static Impl join(By clause) {
    return new Impl(new JoinArguments(clause));
  }
@Test
  @Category(NeedsRunner.class)
  public void testCoGroupByDifferentFields() {
    // Inputs.
    PCollection<Row> pc1 =
        pipeline
            .apply(
                "Create1",
                Create.of(
                    Row.withSchema(CG_SCHEMA_1).addValues("user1", 1, "us").build(),
                    Row.withSchema(CG_SCHEMA_1).addValues("user1", 2, "us").build(),
                    Row.withSchema(CG_SCHEMA_1).addValues("user1", 3, "il").build(),
                    Row.withSchema(CG_SCHEMA_1).addValues("user1", 4, "il").build(),
                    Row.withSchema(CG_SCHEMA_1).addValues("user2", 5, "fr").build(),
                    Row.withSchema(CG_SCHEMA_1).addValues("user2", 6, "fr").build(),
                    Row.withSchema(CG_SCHEMA_1).addValues("user2", 7, "ar").build(),
                    Row.withSchema(CG_SCHEMA_1).addValues("user2", 8, "ar").build()))
            .setRowSchema(CG_SCHEMA_1);
    PCollection<Row> pc2 =
        pipeline
            .apply(
                "Create2",
                Create.of(
                    Row.withSchema(CG_SCHEMA_2).addValues("user1", 9, "us").build(),
                    Row.withSchema(CG_SCHEMA_2).addValues("user1", 10, "us").build(),
                    Row.withSchema(CG_SCHEMA_2).addValues("user1", 11, "il").build(),
                    Row.withSchema(CG_SCHEMA_2).addValues("user1", 12, "il").build(),
                    Row.withSchema(CG_SCHEMA_2).addValues("user2", 13, "fr").build(),
                    Row.withSchema(CG_SCHEMA_2).addValues("user2", 14, "fr").build(),
                    Row.withSchema(CG_SCHEMA_2).addValues("user2", 15, "ar").build(),
                    Row.withSchema(CG_SCHEMA_2).addValues("user2", 16, "ar").build()))
            .setRowSchema(CG_SCHEMA_2);
    PCollection<Row> pc3 =
        pipeline
            .apply(
                "Create3",
                Create.of(
                    Row.withSchema(CG_SCHEMA_3).addValues("user1", 17, "us").build(),
                    Row.withSchema(CG_SCHEMA_3).addValues("user1", 18, "us").build(),
                    Row.withSchema(CG_SCHEMA_3).addValues("user1", 19, "il").build(),
                    Row.withSchema(CG_SCHEMA_3).addValues("user1", 20, "il").build(),
                    Row.withSchema(CG_SCHEMA_3).addValues("user2", 21, "fr").build(),
                    Row.withSchema(CG_SCHEMA_3).addValues("user2", 22, "fr").build(),
                    Row.withSchema(CG_SCHEMA_3).addValues("user2", 23, "ar").build(),
                    Row.withSchema(CG_SCHEMA_3).addValues("user2", 24, "ar").build()))
            .setRowSchema(CG_SCHEMA_3);

    // Expected outputs
    Schema expectedSchema =
        Schema.builder()
            .addRowField("key", SIMPLE_CG_KEY_SCHEMA)
            .addIterableField("pc1", FieldType.row(CG_SCHEMA_1))
            .addIterableField("pc2", FieldType.row(CG_SCHEMA_2))
            .addIterableField("pc3", FieldType.row(CG_SCHEMA_3))
            .build();
    Row key1Joined =
        Row.withSchema(expectedSchema)
            .addValue(Row.withSchema(SIMPLE_CG_KEY_SCHEMA).addValues("user1", "us").build())
            .addIterable(
                Lists.newArrayList(
                    Row.withSchema(CG_SCHEMA_1).addValues("user1", 1, "us").build(),
                    Row.withSchema(CG_SCHEMA_1).addValues("user1", 2, "us").build()))
            .addIterable(
                Lists.newArrayList(
                    Row.withSchema(CG_SCHEMA_2).addValues("user1", 9, "us").build(),
                    Row.withSchema(CG_SCHEMA_2).addValues("user1", 10, "us").build()))
            .addIterable(
                Lists.newArrayList(
                    Row.withSchema(CG_SCHEMA_3).addValues("user1", 17, "us").build(),
                    Row.withSchema(CG_SCHEMA_3).addValues("user1", 18, "us").build()))
            .build();

    Row key2Joined =
        Row.withSchema(expectedSchema)
            .addValue(Row.withSchema(SIMPLE_CG_KEY_SCHEMA).addValues("user1", "il").build())
            .addIterable(
                Lists.newArrayList(
                    Row.withSchema(CG_SCHEMA_1).addValues("user1", 3, "il").build(),
                    Row.withSchema(CG_SCHEMA_1).addValues("user1", 4, "il").build()))
            .addIterable(
                Lists.newArrayList(
                    Row.withSchema(CG_SCHEMA_2).addValues("user1", 11, "il").build(),
                    Row.withSchema(CG_SCHEMA_2).addValues("user1", 12, "il").build()))
            .addIterable(
                Lists.newArrayList(
                    Row.withSchema(CG_SCHEMA_3).addValues("user1", 19, "il").build(),
                    Row.withSchema(CG_SCHEMA_3).addValues("user1", 20, "il").build()))
            .build();

    Row key3Joined =
        Row.withSchema(expectedSchema)
            .addValue(Row.withSchema(SIMPLE_CG_KEY_SCHEMA).addValues("user2", "fr").build())
            .addIterable(
                Lists.newArrayList(
                    Row.withSchema(CG_SCHEMA_1).addValues("user2", 5, "fr").build(),
                    Row.withSchema(CG_SCHEMA_1).addValues("user2", 6, "fr").build()))
            .addIterable(
                Lists.newArrayList(
                    Row.withSchema(CG_SCHEMA_2).addValues("user2", 13, "fr").build(),
                    Row.withSchema(CG_SCHEMA_2).addValues("user2", 14, "fr").build()))
            .addIterable(
                Lists.newArrayList(
                    Row.withSchema(CG_SCHEMA_3).addValues("user2", 21, "fr").build(),
                    Row.withSchema(CG_SCHEMA_3).addValues("user2", 22, "fr").build()))
            .build();

    Row key4Joined =
        Row.withSchema(expectedSchema)
            .addValue(Row.withSchema(SIMPLE_CG_KEY_SCHEMA).addValues("user2", "ar").build())
            .addIterable(
                Lists.newArrayList(
                    Row.withSchema(CG_SCHEMA_1).addValues("user2", 7, "ar").build(),
                    Row.withSchema(CG_SCHEMA_1).addValues("user2", 8, "ar").build()))
            .addIterable(
                Lists.newArrayList(
                    Row.withSchema(CG_SCHEMA_2).addValues("user2", 15, "ar").build(),
                    Row.withSchema(CG_SCHEMA_2).addValues("user2", 16, "ar").build()))
            .addIterable(
                Lists.newArrayList(
                    Row.withSchema(CG_SCHEMA_3).addValues("user2", 23, "ar").build(),
                    Row.withSchema(CG_SCHEMA_3).addValues("user2", 24, "ar").build()))
            .build();

    PCollection<Row> joined1 =
        PCollectionTuple.of("pc1", pc1, "pc2", pc2, "pc3", pc3)
            .apply(
                "CoGroup1",
                CoGroup.join("pc1", By.fieldNames("user", "country"))
                    .join("pc2", By.fieldNames("user2", "country2"))
                    .join("pc3", By.fieldNames("user3", "country3")));
    PCollection<Row> joined2 =
        PCollectionTuple.of("pc1", pc1, "pc2", pc2, "pc3", pc3)
            .apply(
                "CoGroup2",
                CoGroup.join("pc1", By.fieldNames("user", "country"))
                    .join("pc2", By.fieldNames("user2", "country2").withSideInput())
                    .join("pc3", By.fieldNames("user3", "country3")));
    PCollection<Row> joined3 =
        PCollectionTuple.of("pc1", pc1, "pc2", pc2, "pc3", pc3)
            .apply(
                "CoGroup3",
                CoGroup.join("pc1", By.fieldNames("user", "country"))
                    .join("pc2", By.fieldNames("user2", "country2").withSideInput())
                    .join("pc3", By.fieldNames("user3", "country3").withSideInput()));

    List<Row> expected = ImmutableList.of(key1Joined, key2Joined, key3Joined, key4Joined);

    PAssert.that(joined1).satisfies(actual -> containsJoinedFields(expected, actual));
    PAssert.that(joined2).satisfies(actual -> containsJoinedFields(expected, actual));
    PAssert.that(joined3).satisfies(actual -> containsJoinedFields(expected, actual));

    pipeline.run();
  }