public static Type toZetaSqlType(FieldType fieldType) {
    switch (fieldType.getTypeName()) {
      case INT64:
        return TypeFactory.createSimpleType(TypeKind.TYPE_INT64);
      case DOUBLE:
        return TypeFactory.createSimpleType(TypeKind.TYPE_DOUBLE);
      case BOOLEAN:
        return TypeFactory.createSimpleType(TypeKind.TYPE_BOOL);
      case STRING:
        return TypeFactory.createSimpleType(TypeKind.TYPE_STRING);
      case BYTES:
        return TypeFactory.createSimpleType(TypeKind.TYPE_BYTES);
      case DECIMAL:
        return TypeFactory.createSimpleType(TypeKind.TYPE_NUMERIC);
      case DATETIME:
        // TODO[https://github.com/apache/beam/issues/20364]: Mapping TIMESTAMP to a Beam
        // LogicalType instead?
        return TypeFactory.createSimpleType(TypeKind.TYPE_TIMESTAMP);
      case LOGICAL_TYPE:
        String identifier = fieldType.getLogicalType().getIdentifier();
        if (SqlTypes.DATE.getIdentifier().equals(identifier)) {
          return TypeFactory.createSimpleType(TypeKind.TYPE_DATE);
        } else if (SqlTypes.TIME.getIdentifier().equals(identifier)) {
          return TypeFactory.createSimpleType(TypeKind.TYPE_TIME);
        } else if (SqlTypes.DATETIME.getIdentifier().equals(identifier)) {
          return TypeFactory.createSimpleType(TypeKind.TYPE_DATETIME);
        } else {
          throw new UnsupportedOperationException("Unknown Beam logical type: " + identifier);
        }
      case ARRAY:
        return toZetaSqlArrayType(fieldType.getCollectionElementType());
      case ROW:
        return toZetaSqlStructType(fieldType.getRowSchema());
      default:
        throw new UnsupportedOperationException(
            "Unknown Beam fieldType: " + fieldType.getTypeName());
    }
  }
@Test
  public void testBeamFieldTypeToZetaSqlType() {
    assertEquals(ZetaSqlBeamTranslationUtils.toZetaSqlType(TEST_FIELD_TYPE), TEST_TYPE);
  }