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 testFlatSchemaWith2DArrayNestedField() {

    Schema banksSchema = Schema.builder().addStringField("name").addStringField("address").build();
    Schema transactionSchema =
        Schema.builder()
            .addArrayField("banks", Schema.FieldType.row(banksSchema))
            .addDoubleField("purchaseAmount")
            .build();
    Schema nestedSchema =
        Schema.builder()
            .addArrayField("transactions", Schema.FieldType.row(transactionSchema))
            .build();

    String bankName1 = "bank1_1";
    String bankName2 = "bank1_2";
    String bankName3 = "bank2_1";
    String bankName4 = "bank2_2";
    String bankAddress1 = "address1_1";
    String bankAddress2 = "address1_2";
    String bankAddress3 = "address2_1";
    String bankAddress4 = "address2_2";
    double purchaseAmount1 = 1.0;
    double purchaseAmount2 = 2.0;

    Row bank1 = Row.withSchema(banksSchema).addValues(bankName1, bankAddress1).build();
    Row bank2 = Row.withSchema(banksSchema).addValues(bankName2, bankAddress2).build();
    Row bank3 = Row.withSchema(banksSchema).addValues(bankName3, bankAddress3).build();
    Row bank4 = Row.withSchema(banksSchema).addValues(bankName4, bankAddress4).build();
    Row transactionOne =
        Row.withSchema(transactionSchema).addArray(bank1, bank2).addValue(purchaseAmount1).build();
    Row transactionTwo =
        Row.withSchema(transactionSchema).addArray(bank3, bank4).addValue(purchaseAmount2).build();
    Row row = Row.withSchema(nestedSchema).addArray(transactionOne, transactionTwo).build();

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

    Schema expectedUnnestedSchema =
        Schema.builder()
            .addArrayField("transactions_purchaseAmount", FieldType.DOUBLE)
            .addArrayField("transactions_banks_name", FieldType.array(FieldType.STRING))
            .addArrayField("transactions_banks_address", FieldType.array(FieldType.STRING))
            .build();
    assertEquals(expectedUnnestedSchema, unnested.getSchema());

    Row expectedUnnestedRow =
        Row.withSchema(unnested.getSchema())
            .addArray(purchaseAmount1, purchaseAmount2)
            .addArray(Arrays.asList(bankName1, bankName2), Arrays.asList(bankName3, bankName4))
            .addArray(
                Arrays.asList(bankAddress1, bankAddress2),
                Arrays.asList(bankAddress3, bankAddress4))
            .build();
    PAssert.that(unnested).containsInAnyOrder(expectedUnnestedRow);

    pipeline.run();
  }