public static Row toBeamRow(GenericRecord record, Schema schema, ConversionOptions options) {
    List<Object> valuesInOrder =
        schema.getFields().stream()
            .map(
                field -> {
                  try {
                    org.apache.avro.Schema.Field avroField =
                        record.getSchema().getField(field.getName());
                    Object value = avroField != null ? record.get(avroField.pos()) : null;
                    return convertAvroFormat(field.getType(), value, options);
                  } catch (Exception cause) {
                    throw new IllegalArgumentException(
                        "Error converting field " + field + ": " + cause.getMessage(), cause);
                  }
                })
            .collect(toList());

    return Row.withSchema(schema).addValues(valuesInOrder).build();
  }
@Test
  public void testToBeamRow_projection() {
    long testId = 123L;
    // recordSchema is a projection of FLAT_TYPE schema
    org.apache.avro.Schema recordSchema =
        org.apache.avro.SchemaBuilder.record("__root__").fields().optionalLong("id").endRecord();
    GenericData.Record record = new GenericData.Record(recordSchema);
    record.put("id", testId);

    Row expected = Row.withSchema(FLAT_TYPE).withFieldValue("id", testId).build();
    Row actual =
        BigQueryUtils.toBeamRow(
            record, FLAT_TYPE, BigQueryUtils.ConversionOptions.builder().build());
    assertEquals(expected, actual);
  }