public static Struct beamRowToStruct(Row row) {
    Struct.Builder structBuilder = Struct.newBuilder();
    List<Schema.Field> fields = row.getSchema().getFields();
    fields.forEach(
        field -> {
          String column = field.getName();
          switch (field.getType().getTypeName()) {
            case ROW:
              @Nullable Row subRow = row.getRow(column);
              if (subRow == null) {
                structBuilder.set(column).to(beamTypeToSpannerType(field.getType()), null);
              } else {
                structBuilder
                    .set(column)
                    .to(beamTypeToSpannerType(field.getType()), beamRowToStruct(subRow));
              }
              break;
            case ARRAY:
              addIterableToStructBuilder(structBuilder, row.getArray(column), field);
              break;
            case ITERABLE:
              addIterableToStructBuilder(structBuilder, row.getIterable(column), field);
              break;
            case FLOAT:
              structBuilder.set(column).to(row.getFloat(column));
              break;
            case DOUBLE:
              structBuilder.set(column).to(row.getDouble(column));
              break;
            case INT16:
              @Nullable Short int16 = row.getInt16(column);
              if (int16 == null) {
                structBuilder.set(column).to((Long) null);
              } else {
                structBuilder.set(column).to(int16);
              }
              break;
            case INT32:
              @Nullable Integer int32 = row.getInt32(column);
              if (int32 == null) {
                structBuilder.set(column).to((Long) null);
              } else {
                structBuilder.set(column).to(int32);
              }
              break;
            case INT64:
              structBuilder.set(column).to(row.getInt64(column));
              break;
            case DECIMAL:
              @Nullable BigDecimal decimal = row.getDecimal(column);
              // BigDecimal is not nullable
              if (decimal == null) {
                checkNotNull(decimal, "Null decimal at column " + column);
              } else {
                structBuilder.set(column).to(decimal);
              }
              break;
              // TODO: implement logical type date and timestamp
            case DATETIME:
              @Nullable ReadableDateTime dateTime = row.getDateTime(column);
              if (dateTime == null) {
                structBuilder.set(column).to((Timestamp) null);
              } else {
                structBuilder.set(column).to(Timestamp.parseTimestamp(dateTime.toString()));
              }
              break;
            case STRING:
              structBuilder.set(column).to(row.getString(column));
              break;
            case BYTE:
              @Nullable Byte byteValue = row.getByte(column);
              if (byteValue == null) {
                structBuilder.set(column).to((Long) null);
              } else {
                structBuilder.set(column).to(byteValue);
              }
              break;
            case BYTES:
              byte @Nullable [] bytes = row.getBytes(column);
              if (bytes == null) {
                structBuilder.set(column).to((ByteArray) null);
              } else {
                structBuilder.set(column).to(ByteArray.copyFrom(bytes));
              }
              break;
            case BOOLEAN:
              structBuilder.set(column).to(row.getBoolean(column));
              break;
            default:
              throw new IllegalArgumentException(
                  String.format(
                      "Unsupported beam type '%s' while translating row to struct.",
                      field.getType().getTypeName()));
          }
        });
    return structBuilder.build();
  }
@Test
  public void testBeamRowToStruct() {
    Schema schema =
        getSchemaTemplate()
            .addIterableField("f_iterable", Schema.FieldType.INT64)
            .addDecimalField("f_decimal")
            .addInt16Field("f_int16")
            .addInt32Field("f_int32")
            .addByteField("f_byte")
            .addArrayField("f_float_array", Schema.FieldType.FLOAT)
            .addArrayField("f_double_array", Schema.FieldType.DOUBLE)
            .addArrayField("f_decimal_array", Schema.FieldType.DECIMAL)
            .addArrayField("f_boolean_array", Schema.FieldType.BOOLEAN)
            .addArrayField("f_string_array", Schema.FieldType.STRING)
            .addArrayField("f_bytes_array", Schema.FieldType.BYTES)
            .addArrayField("f_datetime_array", Schema.FieldType.DATETIME)
            .build();
    Row row =
        getRowTemplate(schema)
            .withFieldValue("f_iterable", ImmutableList.of(20L))
            .withFieldValue("f_decimal", BigDecimal.ONE)
            .withFieldValue("f_int16", (short) 2)
            .withFieldValue("f_int32", 0x7fffffff)
            .withFieldValue("f_byte", Byte.parseByte("127"))
            .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_boolean_array", ImmutableList.of(false, true))
            .withFieldValue("f_string_array", ImmutableList.of("donald_duck", "micky_mouse"))
            .withFieldValue(
                "f_bytes_array",
                ImmutableList.of("some_bytes".getBytes(UTF_8), "some_bytes".getBytes(UTF_8)))
            .withFieldValue(
                "f_datetime_array",
                ImmutableList.of(
                    DateTime.parse("2077-10-15T00:00:00+00:00"),
                    DateTime.parse("2077-10-15T00:00:00+00:00")))
            .build();
    Struct struct =
        getStructTemplate()
            .set("f_iterable")
            .toInt64Array(ImmutableList.of(20L))
            .set("f_decimal")
            .to(BigDecimal.ONE)
            .set("f_int16")
            .to((short) 2)
            .set("f_int32")
            .to(0x7fffffff)
            .set("f_byte")
            .to(Byte.parseByte("127"))
            .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_boolean_array")
            .toBoolArray(ImmutableList.of(false, true))
            .set("f_string_array")
            .toStringArray(ImmutableList.of("donald_duck", "micky_mouse"))
            .set("f_bytes_array")
            .toBytesArray(
                ImmutableList.of(
                    ByteArray.copyFrom("some_bytes".getBytes(UTF_8)),
                    ByteArray.copyFrom("some_bytes".getBytes(UTF_8))))
            .set("f_datetime_array")
            .toTimestampArray(
                ImmutableList.of(
                    Timestamp.parseTimestamp("2077-10-15T00:00:00Z"),
                    Timestamp.parseTimestamp("2077-10-15T00:00:00Z")))
            .build();
    assertEquals(struct, StructUtils.beamRowToStruct(row));
  }