public static Value toZetaSqlValue(@Nullable Object object, FieldType fieldType) {
    if (object == null) {
      return Value.createNullValue(toZetaSqlType(fieldType));
    }
    switch (fieldType.getTypeName()) {
      case INT64:
        return Value.createInt64Value((Long) object);
      case DOUBLE:
        return Value.createDoubleValue((Double) object);
      case BOOLEAN:
        return Value.createBoolValue((Boolean) object);
      case STRING:
        return Value.createStringValue((String) object);
      case BYTES:
        return Value.createBytesValue(ByteString.copyFrom((byte[]) object));
      case DECIMAL:
        return Value.createNumericValue((BigDecimal) object);
      case DATETIME:
        return Value.createTimestampValueFromUnixMicros(
            LongMath.checkedMultiply(((Instant) object).getMillis(), MICROS_PER_MILLI));
      case LOGICAL_TYPE:
        String identifier = fieldType.getLogicalType().getIdentifier();
        if (SqlTypes.DATE.getIdentifier().equals(identifier)) {
          if (object instanceof Long) { // base type
            return Value.createDateValue(((Long) object).intValue());
          } else { // input type
            return Value.createDateValue((LocalDate) object);
          }
        } else if (SqlTypes.TIME.getIdentifier().equals(identifier)) {
          LocalTime localTime;
          if (object instanceof Long) { // base type
            localTime = LocalTime.ofNanoOfDay((Long) object);
          } else { // input type
            localTime = (LocalTime) object;
          }
          return Value.createTimeValue(localTime);
        } else if (SqlTypes.DATETIME.getIdentifier().equals(identifier)) {
          LocalDateTime datetime;
          if (object instanceof Row) { // base type
            datetime =
                LocalDateTime.of(
                    LocalDate.ofEpochDay(((Row) object).getInt64(DateTime.DATE_FIELD_NAME)),
                    LocalTime.ofNanoOfDay(((Row) object).getInt64(DateTime.TIME_FIELD_NAME)));
          } else { // input type
            datetime = (LocalDateTime) object;
          }
          return Value.createDatetimeValue(datetime);
        } else {
          throw new UnsupportedOperationException("Unknown Beam logical type: " + identifier);
        }
      case ARRAY:
        return toZetaSqlArrayValue((List<Object>) object, fieldType.getCollectionElementType());
      case ROW:
        return toZetaSqlStructValue((Row) object, fieldType.getRowSchema());
      default:
        throw new UnsupportedOperationException(
            "Unknown Beam fieldType: " + fieldType.getTypeName());
    }
  }
@Test
  public void testJavaObjectToZetaSqlValue() {
    assertEquals(ZetaSqlBeamTranslationUtils.toZetaSqlValue(TEST_ROW, TEST_FIELD_TYPE), TEST_VALUE);
  }