public static <T> Inner<T> create() {
    return new Inner<>();
  }
@Test
  @Category(NeedsRunner.class)
  public void addNestedMapField() {
    Schema nested = Schema.builder().addStringField("field1").build();
    Schema schema =
        Schema.builder()
            .addMapField("map", Schema.FieldType.STRING, Schema.FieldType.row(nested))
            .build();

    Row subRow = Row.withSchema(nested).addValue("value").build();
    Row row = Row.withSchema(schema).addValue(ImmutableMap.of("key", subRow)).build();
    PCollection<Row> added =
        pipeline
            .apply(Create.of(row).withRowSchema(schema))
            .apply(
                AddFields.<Row>create()
                    .field("map.field2", Schema.FieldType.INT32)
                    .field("map.field3", Schema.FieldType.array(Schema.FieldType.STRING))
                    .field("map.field4", Schema.FieldType.iterable(Schema.FieldType.STRING)));

    Schema expectedNestedSchema =
        Schema.builder()
            .addStringField("field1")
            .addNullableField("field2", Schema.FieldType.INT32)
            .addNullableField("field3", Schema.FieldType.array(Schema.FieldType.STRING))
            .addNullableField("field4", Schema.FieldType.iterable(Schema.FieldType.STRING))
            .build();
    Schema expectedSchema =
        Schema.builder()
            .addMapField("map", Schema.FieldType.STRING, Schema.FieldType.row(expectedNestedSchema))
            .build();
    assertEquals(expectedSchema, added.getSchema());

    Row expectedNested =
        Row.withSchema(expectedNestedSchema).addValues("value", null, null, null).build();
    Row expected =
        Row.withSchema(expectedSchema).addValue(ImmutableMap.of("key", expectedNested)).build();

    PAssert.that(added).containsInAnyOrder(expected);
    pipeline.run();
  }