public static <T> ByFields<T> byFieldNames(String... fieldNames) {
    return ByFields.of(FieldAccessDescriptor.withFieldNames(fieldNames));
  }
@Test
  @Category(NeedsRunner.class)
  public void testGroupByNestedKey() throws NoSuchSchemaException {
    PCollection<Row> grouped =
        pipeline
            .apply(
                Create.of(
                    Outer.of(Basic.of("key1", 1L, "value1")),
                    Outer.of(Basic.of("key1", 1L, "value2")),
                    Outer.of(Basic.of("key2", 2L, "value3")),
                    Outer.of(Basic.of("key2", 2L, "value4"))))
            .apply(Group.byFieldNames("inner.field1", "inner.field2"));

    Schema keySchema = Schema.builder().addStringField("field1").addInt64Field("field2").build();
    Schema outputSchema =
        Schema.builder()
            .addRowField("key", keySchema)
            .addIterableField("value", FieldType.row(OUTER_SCHEMA))
            .build();

    SerializableFunction<Outer, Row> toRow =
        pipeline.getSchemaRegistry().getToRowFunction(Outer.class);

    List<Row> expected =
        ImmutableList.of(
            Row.withSchema(outputSchema)
                .addValue(Row.withSchema(keySchema).addValues("key1", 1L).build())
                .addIterable(
                    ImmutableList.of(
                        toRow.apply(Outer.of(Basic.of("key1", 1L, "value1"))),
                        toRow.apply(Outer.of(Basic.of("key1", 1L, "value2")))))
                .build(),
            Row.withSchema(outputSchema)
                .addValue(Row.withSchema(keySchema).addValues("key2", 2L).build())
                .addIterable(
                    ImmutableList.of(
                        toRow.apply(Outer.of(Basic.of("key2", 2L, "value3"))),
                        toRow.apply(Outer.of(Basic.of("key2", 2L, "value4")))))
                .build());

    PAssert.that(grouped).satisfies(actual -> containsKIterableVs(expected, actual));
    pipeline.run();
  }