public static <T> ByFields<T> byFieldNames(String... fieldNames) {
    return ByFields.of(FieldAccessDescriptor.withFieldNames(fieldNames));
  }
@Test
  @Category(NeedsRunner.class)
  public void testByKeyWithSchemaAggregateFn() {
    Collection<Aggregate> elements =
        ImmutableList.of(
            Aggregate.of(1, 1, 2),
            Aggregate.of(2, 1, 3),
            Aggregate.of(3, 2, 4),
            Aggregate.of(4, 2, 5));

    PCollection<Row> aggregations =
        pipeline
            .apply(Create.of(elements))
            .apply(
                Group.<Aggregate>byFieldNames("field2")
                    .aggregateField("field1", Sum.ofLongs(), "field1_sum")
                    .aggregateField("field3", Sum.ofIntegers(), "field3_sum")
                    .aggregateField("field1", Top.largestLongsFn(1), "field1_top"));

    Schema keySchema = Schema.builder().addInt64Field("field2").build();
    Schema valueSchema =
        Schema.builder()
            .addInt64Field("field1_sum")
            .addInt32Field("field3_sum")
            .addArrayField("field1_top", FieldType.INT64)
            .build();
    Schema outputSchema =
        Schema.builder().addRowField("key", keySchema).addRowField("value", valueSchema).build();

    List<Row> expected =
        ImmutableList.of(
            Row.withSchema(outputSchema)
                .addValue(Row.withSchema(keySchema).addValue(1L).build())
                .addValue(Row.withSchema(valueSchema).addValue(3L).addValue(5).addArray(2L).build())
                .build(),
            Row.withSchema(outputSchema)
                .addValue(Row.withSchema(keySchema).addValue(2L).build())
                .addValue(Row.withSchema(valueSchema).addValue(7L).addValue(9).addArray(4L).build())
                .build());
    PAssert.that(aggregations).satisfies(actual -> containsKvRows(expected, actual));

    pipeline.run();
  }