public static <T> PTransform<PCollection<T>, PCollection<T>> intersectDistinct(
      PCollection<T> rightCollection) {
    checkNotNull(rightCollection, "rightCollection argument is null");
    return new SetImpl<>(rightCollection, intersectDistinct());
  }
@Test
  @Category(NeedsRunner.class)
  public void testIntersectionCollectionList() {

    PCollection<String> third = p.apply("third", Create.of(Arrays.asList("b", "b", "c", "f")));
    PCollection<Row> thirdRows = p.apply("thirdRows", Create.of(toRows("b", "b", "c", "f")));

    PAssert.that(
            PCollectionList.of(first)
                .and(second)
                .and(third)
                .apply("stringsCols", Sets.intersectDistinct()))
        .containsInAnyOrder("b", "c");

    PCollection<Row> results =
        PCollectionList.of(firstRows)
            .and(secondRows)
            .and(thirdRows)
            .apply("rowCols", Sets.intersectDistinct());

    PAssert.that(results).containsInAnyOrder(toRows("b", "c"));

    assertEquals(schema, results.getSchema());

    p.run();
  }