public static FieldType fieldTypeForJavaType(TypeDescriptor typeDescriptor) {
    // TODO: Convert for registered logical types.
    if (typeDescriptor.isArray()
        || typeDescriptor.isSubtypeOf(TypeDescriptor.of(Collection.class))) {
      return getArrayFieldType(typeDescriptor);
    } else if (typeDescriptor.isSubtypeOf(TypeDescriptor.of(Map.class))) {
      return getMapFieldType(typeDescriptor);
    } else if (typeDescriptor.isSubtypeOf(TypeDescriptor.of(Iterable.class))) {
      return getIterableFieldType(typeDescriptor);
    } else if (typeDescriptor.isSubtypeOf(TypeDescriptor.of(Row.class))) {
      throw new IllegalArgumentException(
          "Cannot automatically determine a field type from a Row class"
              + " as we cannot determine the schema. You should set a field type explicitly.");
    } else {
      TypeName typeName = PRIMITIVE_MAPPING.inverse().get(typeDescriptor);
      if (typeName == null) {
        throw new RuntimeException("Couldn't find field type for " + typeDescriptor);
      }
      return FieldType.of(typeName);
    }
  }
@Test
  public void testPrimitiveTypeToFieldType() {
    assertEquals(
        FieldType.BYTE, FieldTypeDescriptors.fieldTypeForJavaType(TypeDescriptors.bytes()));
    assertEquals(
        FieldType.INT16, FieldTypeDescriptors.fieldTypeForJavaType(TypeDescriptors.shorts()));
    assertEquals(
        FieldType.INT32, FieldTypeDescriptors.fieldTypeForJavaType(TypeDescriptors.integers()));
    assertEquals(
        FieldType.INT64, FieldTypeDescriptors.fieldTypeForJavaType(TypeDescriptors.longs()));
    assertEquals(
        FieldType.DECIMAL,
        FieldTypeDescriptors.fieldTypeForJavaType(TypeDescriptors.bigdecimals()));
    assertEquals(
        FieldType.FLOAT, FieldTypeDescriptors.fieldTypeForJavaType(TypeDescriptors.floats()));
    assertEquals(
        FieldType.DOUBLE, FieldTypeDescriptors.fieldTypeForJavaType(TypeDescriptors.doubles()));
    assertEquals(
        FieldType.STRING, FieldTypeDescriptors.fieldTypeForJavaType(TypeDescriptors.strings()));
    assertEquals(
        FieldType.DATETIME,
        FieldTypeDescriptors.fieldTypeForJavaType(TypeDescriptor.of(Instant.class)));
    assertEquals(
        FieldType.BOOLEAN, FieldTypeDescriptors.fieldTypeForJavaType(TypeDescriptors.booleans()));
    assertEquals(
        FieldType.BYTES,
        FieldTypeDescriptors.fieldTypeForJavaType(TypeDescriptor.of(byte[].class)));
  }