class JdbcUtil {

  static JdbcIO.PreparedStatementSetCaller getPreparedStatementSetCaller(
      Schema.FieldType fieldType) {
    switch (fieldType.getTypeName()) {
      case ARRAY:
      case ITERABLE:
        return (element, ps, i, fieldWithIndex) ->
            ps.setArray(
                i + 1,
                ps.getConnection()
                    .createArrayOf(
                        fieldType.getCollectionElementType().getTypeName().name(),
                        element.getArray(fieldWithIndex.getIndex()).toArray()));
      case LOGICAL_TYPE:
        {
          String logicalTypeName = fieldType.getLogicalType().getIdentifier();
          JDBCType jdbcType = JDBCType.valueOf(logicalTypeName);
          switch (jdbcType) {
            case DATE:
              return (element, ps, i, fieldWithIndex) ->
                  ps.setDate(
                      i + 1,
                      new Date(
                          getDateOrTimeOnly(
                                  element.getDateTime(fieldWithIndex.getIndex()).toDateTime(), true)
                              .getTime()
                              .getTime()));
            case TIME:
              return (element, ps, i, fieldWithIndex) ->
                  ps.setTime(
                      i + 1,
                      new Time(
                          getDateOrTimeOnly(
                                  element.getDateTime(fieldWithIndex.getIndex()).toDateTime(),
                                  false)
                              .getTime()
                              .getTime()));
            case TIMESTAMP_WITH_TIMEZONE:
              return (element, ps, i, fieldWithIndex) -> {
                Calendar calendar =
                    withTimestampAndTimezone(
                        element.getDateTime(fieldWithIndex.getIndex()).toDateTime());
                ps.setTimestamp(i + 1, new Timestamp(calendar.getTime().getTime()), calendar);
              };
            default:
              return getPreparedStatementSetCaller(fieldType.getLogicalType().getBaseType());
          }
        }
      default:
        {
          if (typeNamePsSetCallerMap.containsKey(fieldType.getTypeName())) {
            return typeNamePsSetCallerMap.get(fieldType.getTypeName());
          } else {
            throw new RuntimeException(
                fieldType.getTypeName().name()
                    + " in schema is not supported while writing. Please provide statement and preparedStatementSetter");
          }
        }
    }
  }

}