@SuppressWarnings({
    "nullness" // TODO(https://github.com/apache/beam/issues/20497)
  })
  public static Row structToBeamRow(Struct struct, Schema schema) {
    Map<String, @Nullable Object> structValues =
        schema.getFields().stream()
            .collect(
                HashMap::new,
                (map, field) -> map.put(field.getName(), getStructValue(struct, field)),
                Map::putAll);
    return Row.withSchema(schema).withFieldValues(structValues).build();
  }
@Test
  public void testStructToBeamRow() {
    Schema schema =
        getSchemaTemplate()
            .addDateTimeField("f_date")
            .addDecimalField("f_decimal")
            .addArrayField("f_boolean_array", Schema.FieldType.BOOLEAN)
            .addArrayField("f_string_array", Schema.FieldType.STRING)
            .addArrayField("f_float_array", Schema.FieldType.FLOAT)
            .addArrayField("f_double_array", Schema.FieldType.DOUBLE)
            .addArrayField("f_decimal_array", Schema.FieldType.DECIMAL)
            .addArrayField("f_date_array", Schema.FieldType.DATETIME)
            .addArrayField("f_timestamp_array", Schema.FieldType.DATETIME)
            .build();

    Row row =
        getRowTemplate(schema)
            .withFieldValue("f_date", DateTime.parse("2077-10-24"))
            .withFieldValue("f_decimal", BigDecimal.valueOf(Long.MIN_VALUE))
            .withFieldValue("f_boolean_array", ImmutableList.of(false, true))
            .withFieldValue("f_string_array", ImmutableList.of("donald_duck", "micky_mouse"))
            .withFieldValue("f_float_array", ImmutableList.of(3.0f, 4.0f))
            .withFieldValue("f_double_array", ImmutableList.of(1., 2.))
            .withFieldValue(
                "f_decimal_array",
                ImmutableList.of(
                    BigDecimal.valueOf(Long.MIN_VALUE), BigDecimal.valueOf(Long.MAX_VALUE)))
            .withFieldValue(
                "f_date_array",
                ImmutableList.of(DateTime.parse("2077-10-24"), DateTime.parse("2077-10-24")))
            .withFieldValue(
                "f_timestamp_array",
                ImmutableList.of(DateTime.parse("2077-01-10"), DateTime.parse("2077-01-10")))
            .build();
    Struct struct =
        getStructTemplate()
            .set("f_date")
            .to(Date.fromYearMonthDay(2077, 10, 24))
            .set("f_decimal")
            .to(BigDecimal.valueOf(Long.MIN_VALUE))
            .set("f_boolean_array")
            .toBoolArray(ImmutableList.of(false, true))
            .set("f_string_array")
            .toStringArray(ImmutableList.of("donald_duck", "micky_mouse"))
            .set("f_float_array")
            .toFloat32Array(ImmutableList.of(3.0f, 4.0f))
            .set("f_double_array")
            .toFloat64Array(ImmutableList.of(1., 2.))
            .set("f_decimal_array")
            .toNumericArray(
                ImmutableList.of(
                    BigDecimal.valueOf(Long.MIN_VALUE), BigDecimal.valueOf(Long.MAX_VALUE)))
            .set("f_date_array")
            .toDateArray(
                ImmutableList.of(
                    Date.fromYearMonthDay(2077, 10, 24), Date.fromYearMonthDay(2077, 10, 24)))
            .set("f_timestamp_array")
            .toTimestampArray(
                ImmutableList.of(
                    Timestamp.ofTimeMicroseconds(
                        DateTime.parse("2077-01-10").toInstant().getMillis() * 1000L),
                    Timestamp.ofTimeMicroseconds(
                        DateTime.parse("2077-01-10").toInstant().getMillis() * 1000L)))
            .build();
    assertEquals(row, StructUtils.structToBeamRow(struct, schema));
  }