public static Impl join(By clause) {
    return new Impl(new JoinArguments(clause));
  }
@Test
  @Category(NeedsRunner.class)
  public void testPartialOuterJoin() {
    List<Row> pc1Rows =
        Lists.newArrayList(
            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());

    List<Row> pc2Rows =
        Lists.newArrayList(
            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("user3", 7, "ar").build());

    List<Row> pc3Rows =
        Lists.newArrayList(
            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(),
            Row.withSchema(CG_SCHEMA_3).addValues("user3", 25, "ar").build());

    PCollection<Row> pc1 = pipeline.apply("Create1", Create.of(pc1Rows)).setRowSchema(CG_SCHEMA_1);
    PCollection<Row> pc2 = pipeline.apply("Create2", Create.of(pc2Rows)).setRowSchema(CG_SCHEMA_2);
    PCollection<Row> pc3 = pipeline.apply("Create3", Create.of(pc3Rows)).setRowSchema(CG_SCHEMA_3);

    // Partial outer join. Missing entries in the "pc2" PCollection will be filled in with nulls,
    // but not others.
    Schema expectedSchema =
        Schema.builder()
            .addField("pc1", FieldType.row(CG_SCHEMA_1))
            .addNullableField("pc2", FieldType.row(CG_SCHEMA_2))
            .addField("pc3", FieldType.row(CG_SCHEMA_3))
            .build();

    PCollection<Row> joined =
        PCollectionTuple.of("pc1", pc1, "pc2", pc2, "pc3", pc3)
            .apply(
                "CoGroup",
                CoGroup.join("pc1", By.fieldNames("user", "country"))
                    .join("pc2", By.fieldNames("user2", "country2").withOptionalParticipation())
                    .join("pc3", By.fieldNames("user3", "country3"))
                    .crossProductJoin());
    assertEquals(expectedSchema, joined.getSchema());

    List<Row> expectedJoinedRows =
        JoinTestUtils.innerJoin(
            pc1Rows,
            pc2Rows,
            pc3Rows,
            new String[] {"user", "country"},
            new String[] {"user2", "country2"},
            new String[] {"user3", "country3"},
            expectedSchema);

    // Manually add the outer-join rows to the list of expected results. Missing results from the
    // middle (pc2) PCollection are filled in with nulls. Missing events from other PCollections
    // are not. Events with key ("user2", "ar) show up in pc1 and pc3 but not in pc2, so we expect
    // the outer join to still produce those rows, with nulls for pc2. Events with key
    // ("user3", "ar) however show up in p2 and pc3, but not in pc1; since pc1 is marked for
    // full participation (no outer join), these events should not be included in the join.
    expectedJoinedRows.add(
        Row.withSchema(expectedSchema)
            .addValues(
                Row.withSchema(CG_SCHEMA_1).addValues("user2", 7, "ar").build(),
                null,
                Row.withSchema(CG_SCHEMA_3).addValues("user2", 23, "ar").build())
            .build());
    expectedJoinedRows.add(
        Row.withSchema(expectedSchema)
            .addValues(
                Row.withSchema(CG_SCHEMA_1).addValues("user2", 7, "ar").build(),
                null,
                Row.withSchema(CG_SCHEMA_3).addValues("user2", 24, "ar").build())
            .build());
    expectedJoinedRows.add(
        Row.withSchema(expectedSchema)
            .addValues(
                Row.withSchema(CG_SCHEMA_1).addValues("user2", 8, "ar").build(),
                null,
                Row.withSchema(CG_SCHEMA_3).addValues("user2", 23, "ar").build())
            .build());
    expectedJoinedRows.add(
        Row.withSchema(expectedSchema)
            .addValues(
                Row.withSchema(CG_SCHEMA_1).addValues("user2", 8, "ar").build(),
                null,
                Row.withSchema(CG_SCHEMA_3).addValues("user2", 24, "ar").build())
            .build());

    PAssert.that(joined).containsInAnyOrder(expectedJoinedRows);
    pipeline.run();
  }