@Override
  protected @UnknownKeyFor @NonNull @Initialized SchemaTransform from(
      JdbcWriteSchemaTransformConfiguration configuration) {
    configuration.validate();
    return new JdbcWriteSchemaTransform(configuration);
  }
@Test
  public void testWriteToTableWithJdbcTypeSpecified() throws SQLException {
    JdbcWriteSchemaTransformProvider provider = null;
    for (SchemaTransformProvider p : ServiceLoader.load(SchemaTransformProvider.class)) {
      if (p instanceof JdbcWriteSchemaTransformProvider) {
        provider = (JdbcWriteSchemaTransformProvider) p;
        break;
      }
    }
    assertNotNull(provider);

    Schema schema =
        Schema.of(
            Schema.Field.of("id", Schema.FieldType.INT64),
            Schema.Field.of("name", Schema.FieldType.STRING));

    List<Row> rows =
        ImmutableList.of(
            Row.withSchema(schema).attachValues(1L, "name1"),
            Row.withSchema(schema).attachValues(2L, "name2"));

    PCollectionRowTuple.of("input", pipeline.apply(Create.of(rows).withRowSchema(schema)))
        .apply(
            provider.from(
                JdbcWriteSchemaTransformProvider.JdbcWriteSchemaTransformConfiguration.builder()
                    .setJdbcUrl(DATA_SOURCE_CONFIGURATION.getUrl().get())
                    .setJdbcType("derby")
                    .setLocation(writeTableName)
                    .build()));
    pipeline.run();
    DatabaseTestHelper.assertRowCount(DATA_SOURCE, writeTableName, 2);
  }