@Override
  protected SchemaTransform from(Configuration configuration) {
    return new ExplodeTransform(configuration);
  }
@Test
  @Category(NeedsRunner.class)
  public void testCrossProduct() {
    PCollection<Row> input = pipeline.apply(Create.of(INPUT_ROWS)).setRowSchema(INPUT_SCHEMA);

    PCollection<Row> exploded =
        PCollectionRowTuple.of(JavaExplodeTransformProvider.INPUT_ROWS_TAG, input)
            .apply(
                new JavaExplodeTransformProvider()
                    .from(
                        JavaExplodeTransformProvider.Configuration.builder()
                            .setFields(ImmutableList.of("a", "c"))
                            .setCrossProduct(true)
                            .build()))
            .get(JavaExplodeTransformProvider.OUTPUT_ROWS_TAG);

    PAssert.that(exploded)
        .containsInAnyOrder(
            Row.withSchema(OUTPUT_SCHEMA).addValues(1, 1.5, "x").build(),
            Row.withSchema(OUTPUT_SCHEMA).addValues(2, 1.5, "x").build(),
            Row.withSchema(OUTPUT_SCHEMA).addValues(1, 1.5, "y").build(),
            Row.withSchema(OUTPUT_SCHEMA).addValues(2, 1.5, "y").build());

    pipeline.run();
  }