@Override
  protected SchemaTransform from(BigQueryFileLoadsWriteSchemaTransformConfiguration configuration) {
    return new BigQueryWriteSchemaTransform(configuration);
  }
@Test
  public void testToWrite() {
    List<
            Pair<
                BigQueryFileLoadsWriteSchemaTransformConfiguration.Builder,
                BigQueryIO.Write<TableRow>>>
        cases =
            Arrays.asList(
                Pair.of(
                    BigQueryFileLoadsWriteSchemaTransformConfiguration.builder()
                        .setTableSpec(BigQueryHelpers.toTableSpec(TABLE_REFERENCE))
                        .setCreateDisposition(CreateDisposition.CREATE_NEVER.name())
                        .setWriteDisposition(WriteDisposition.WRITE_EMPTY.name()),
                    BigQueryIO.writeTableRows()
                        .to(TABLE_REFERENCE)
                        .withCreateDisposition(CreateDisposition.CREATE_NEVER)
                        .withWriteDisposition(WriteDisposition.WRITE_EMPTY)
                        .withSchema(TABLE_SCHEMA)),
                Pair.of(
                    BigQueryFileLoadsWriteSchemaTransformConfiguration.builder()
                        .setTableSpec(BigQueryHelpers.toTableSpec(TABLE_REFERENCE))
                        .setCreateDisposition(CreateDisposition.CREATE_IF_NEEDED.name())
                        .setWriteDisposition(WriteDisposition.WRITE_TRUNCATE.name()),
                    BigQueryIO.writeTableRows()
                        .to(TABLE_REFERENCE)
                        .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
                        .withWriteDisposition(WriteDisposition.WRITE_TRUNCATE)
                        .withSchema(TABLE_SCHEMA)));
    for (Pair<
            BigQueryFileLoadsWriteSchemaTransformConfiguration.Builder, BigQueryIO.Write<TableRow>>
        caze : cases) {
      BigQueryWriteSchemaTransform transform = transformFrom(caze.getLeft().build());
      Map<Identifier, Item> gotDisplayData = DisplayData.from(transform.toWrite(SCHEMA)).asMap();
      Map<Identifier, Item> wantDisplayData = DisplayData.from(caze.getRight()).asMap();
      Set<Identifier> keys = new HashSet<>();
      keys.addAll(gotDisplayData.keySet());
      keys.addAll(wantDisplayData.keySet());
      for (Identifier key : keys) {
        Item got = null;
        Item want = null;
        if (gotDisplayData.containsKey(key)) {
          got = gotDisplayData.get(key);
        }
        if (wantDisplayData.containsKey(key)) {
          want = wantDisplayData.get(key);
        }
        assertEquals(want, got);
      }
    }
  }