public static <T> Flattened<T> flattenedSchema() {
    return new AutoValue_Select_Flattened.Builder<T>()
        .setNameFn(CONCAT_FIELD_NAMES)
        .setNameOverrides(Collections.emptyMap())
        .build();
  }
@Test
  @Category(NeedsRunner.class)
  public void testFlatSchemaWithArrayNestedField() {

    Schema shippingAddressSchema =
        Schema.builder().addStringField("streetAddress").addStringField("city").build();
    Schema transactionSchema =
        Schema.builder()
            .addArrayField("banks", FieldType.STRING)
            .addDoubleField("purchaseAmount")
            .build();
    Schema nestedSchema =
        Schema.builder()
            .addStringField("userId")
            .addRowField("shippingAddress", shippingAddressSchema)
            .addArrayField("transactions", Schema.FieldType.row(transactionSchema))
            .build();

    String userId = "user";
    String street = "street";
    String city = "city";
    String bank1 = "bank1_1";
    String bank2 = "bank1_2";
    String bank3 = "bank2_1";
    String bank4 = "bank2_2";
    double purchaseAmount1 = 1.0;
    double purchaseAmount2 = 2.0;

    Row transactionOne =
        Row.withSchema(transactionSchema).addArray(bank1, bank2).addValue(purchaseAmount1).build();
    Row transactionTwo =
        Row.withSchema(transactionSchema).addArray(bank3, bank4).addValue(purchaseAmount2).build();
    Row address = Row.withSchema(shippingAddressSchema).addValues(street, city).build();
    Row row =
        Row.withSchema(nestedSchema)
            .addValues(userId, address)
            .addArray(transactionOne, transactionTwo)
            .build();

    PCollection<Row> unnested =
        pipeline.apply(Create.of(row).withRowSchema(nestedSchema)).apply(Select.flattenedSchema());

    Schema expectedUnnestedSchema =
        Schema.builder()
            .addStringField("userId")
            .addStringField("shippingAddress_streetAddress")
            .addStringField("shippingAddress_city")
            .addArrayField("transactions_banks", FieldType.array(FieldType.STRING))
            .addArrayField("transactions_purchaseAmount", FieldType.DOUBLE)
            .build();
    assertEquals(expectedUnnestedSchema, unnested.getSchema());

    Row expectedUnnestedRow =
        Row.withSchema(unnested.getSchema())
            .addValues(userId, street, city)
            .addArray(Arrays.asList(bank1, bank2), Arrays.asList(bank3, bank4))
            .addArray(purchaseAmount1, purchaseAmount2)
            .build();
    PAssert.that(unnested).containsInAnyOrder(expectedUnnestedRow);

    pipeline.run();
  }