@VisibleForTesting
  static TableSchema protoTableSchemaFromBeamSchema(Schema schema) {
    Preconditions.checkState(schema.getFieldCount() > 0);

    TableSchema.Builder builder = TableSchema.newBuilder();
    for (Field field : schema.getFields()) {
      builder.addFields(fieldDescriptorFromBeamField(field));
    }
    return builder.build();
  }
@Test
  public void testDescriptorFromSchema() {
    DescriptorProto descriptor =
        TableRowToStorageApiProto.descriptorSchemaFromTableSchema(
            BeamRowToStorageApiProto.protoTableSchemaFromBeamSchema(BASE_SCHEMA), true, false);
    Map<String, Type> types =
        descriptor.getFieldList().stream()
            .collect(
                Collectors.toMap(FieldDescriptorProto::getName, FieldDescriptorProto::getType));
    Map<String, Type> expectedTypes =
        BASE_SCHEMA_PROTO.getFieldList().stream()
            .collect(
                Collectors.toMap(FieldDescriptorProto::getName, FieldDescriptorProto::getType));
    assertEquals(expectedTypes, types);

    Map<String, String> nameMapping =
        BASE_SCHEMA.getFields().stream()
            .collect(Collectors.toMap(f -> f.getName().toLowerCase(), Field::getName));
    descriptor
        .getFieldList()
        .forEach(
            p -> {
              FieldType schemaFieldType =
                  BASE_SCHEMA.getField(nameMapping.get(p.getName())).getType();
              Label label =
                  schemaFieldType.getTypeName().isCollectionType()
                      ? Label.LABEL_REPEATED
                      : schemaFieldType.getNullable() ? Label.LABEL_OPTIONAL : Label.LABEL_REQUIRED;
              assertEquals(label, p.getLabel());
            });
  }