@Override
  protected SchemaTransform from(Configuration configuration) {
    return new JavaMapToFieldsTransform(configuration);
  }
@Test
  @Category(NeedsRunner.class)
  public void testRenameFields() {
    Schema inputSchema =
        Schema.of(
            Schema.Field.of("a", Schema.FieldType.STRING),
            Schema.Field.of("b", Schema.FieldType.INT32),
            Schema.Field.of("c", Schema.FieldType.DOUBLE));

    PCollection<Row> input =
        pipeline
            .apply(
                Create.of(
                    Row.withSchema(inputSchema).addValues("foo", 2, 0.5).build(),
                    Row.withSchema(inputSchema).addValues("bar", 4, 0.25).build()))
            .setRowSchema(inputSchema);

    PCollection<Row> renamed =
        PCollectionRowTuple.of(JavaMapToFieldsTransformProvider.INPUT_ROWS_TAG, input)
            .apply(
                new JavaMapToFieldsTransformProvider()
                    .from(
                        JavaMapToFieldsTransformProvider.Configuration.builder()
                            .setFields(
                                ImmutableMap.of(
                                    "newC",
                                    JavaRowUdf.Configuration.builder().setExpression("c").build(),
                                    "newA",
                                    JavaRowUdf.Configuration.builder().setExpression("a").build()))
                            .build()))
            .get(JavaMapToFieldsTransformProvider.OUTPUT_ROWS_TAG);

    Schema outputSchema = renamed.getSchema();

    PAssert.that(renamed)
        .containsInAnyOrder(
            Row.withSchema(outputSchema)
                .withFieldValue("newC", 0.5)
                .withFieldValue("newA", "foo")
                .build(),
            Row.withSchema(outputSchema)
                .withFieldValue("newC", 0.25)
                .withFieldValue("newA", "bar")
                .build());

    pipeline.run();
  }