@Override
  protected SchemaTransform from(Configuration configuration) {
    return new JavaFilterTransform(configuration);
  }
@Test
  @Category(NeedsRunner.class)
  public void testFilter() {
    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(JavaFilterTransformProvider.INPUT_ROWS_TAG, input)
            .apply(
                new JavaFilterTransformProvider()
                    .from(
                        JavaFilterTransformProvider.Configuration.builder()
                            .setKeep(
                                JavaRowUdf.Configuration.builder()
                                    .setExpression("b + c > 3")
                                    .build())
                            .build()))
            .get(JavaFilterTransformProvider.OUTPUT_ROWS_TAG);

    PAssert.that(renamed)
        .containsInAnyOrder(
            Row.withSchema(inputSchema)
                .withFieldValue("a", "bar")
                .withFieldValue("b", 4)
                .withFieldValue("c", 0.25)
                .build());

    pipeline.run();
  }