public static Impl join(By clause) {
    return new Impl(new JoinArguments(clause));
  }
@Test
  @Category(NeedsRunner.class)
  public void testPojo() {
    List<CgPojo> pc1Rows =
        Lists.newArrayList(
            new CgPojo("user1", 1, "us"),
            new CgPojo("user1", 2, "us"),
            new CgPojo("user1", 3, "il"),
            new CgPojo("user1", 4, "il"));

    List<CgPojo> pc2Rows =
        Lists.newArrayList(
            new CgPojo("user1", 3, "us"),
            new CgPojo("user1", 4, "us"),
            new CgPojo("user1", 5, "il"),
            new CgPojo("user1", 6, "il"));

    PCollection<CgPojo> pc1 = pipeline.apply("Create1", Create.of(pc1Rows));
    PCollection<CgPojo> pc2 = pipeline.apply("Create2", Create.of(pc2Rows));

    PCollection<Row> joined =
        PCollectionTuple.of("pc1", pc1)
            .and("pc2", pc2)
            .apply(
                CoGroup.join("pc1", By.fieldNames("user", "country"))
                    .join("pc2", By.fieldNames("user", "country")));

    Schema expectedSchema =
        Schema.builder()
            .addRowField("key", SIMPLE_CG_KEY_SCHEMA)
            .addIterableField("pc1", FieldType.row(CG_SCHEMA_1))
            .addIterableField("pc2", FieldType.row(CG_SCHEMA_1))
            .build();

    List<Row> expected =
        Lists.newArrayList(
            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_1).addValues("user1", 3, "us").build(),
                        Row.withSchema(CG_SCHEMA_1).addValues("user1", 4, "us").build()))
                .build(),
            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_1).addValues("user1", 5, "il").build(),
                        Row.withSchema(CG_SCHEMA_1).addValues("user1", 6, "il").build()))
                .build());

    assertEquals(expectedSchema, joined.getSchema());
    PAssert.that(joined).satisfies(actual -> containsJoinedFields(expected, actual));

    pipeline.run();
  }