public static <T> ByFields<T> byFieldNames(String... fieldNames) {
    return ByFields.of(FieldAccessDescriptor.withFieldNames(fieldNames));
  }
@Test
  @Category(NeedsRunner.class)
  public void testOutputCoders() {
    Schema keySchema = Schema.builder().addStringField("field1").build();
    Schema outputSchema =
        Schema.builder()
            .addRowField("key", keySchema)
            .addIterableField("value", FieldType.row(BASIC_SCHEMA))
            .build();

    PCollection<Row> grouped =
        pipeline
            .apply(Create.of(Basic.of("key1", 1, "value1")))
            .apply(Group.byFieldNames("field1"));

    assertTrue(grouped.getSchema().equivalent(outputSchema));

    pipeline.run();
  }