protected GenericRowData generateRecord(RowType rowType, Map<String, String> fieldMap) {
        String[] fieldNames = rowType.getFieldNames().toArray(new String[0]);
        int arity = fieldNames.length;
        GenericRowData record = new GenericRowData(arity);
        for (int i = 0; i < arity; i++) {
            String fieldName = fieldNames[i];
            String fieldValue = fieldMap.get(fieldName);
            if (StringUtils.isBlank(fieldValue)) {
                record.setField(i, null);
            } else {
                switch (rowType.getFields().get(i).getType().getTypeRoot()) {
                    case BIGINT:
                        record.setField(i, Long.valueOf(fieldValue));
                        break;
                    case BOOLEAN:
                        record.setField(i, Boolean.valueOf(fieldValue));
                        break;
                    case DOUBLE:
                    case DECIMAL:
                        record.setField(i, Double.valueOf(fieldValue));
                        break;
                    case TIME_WITHOUT_TIME_ZONE:
                    case INTERVAL_DAY_TIME:
                        TimestampData timestampData = TimestampData.fromEpochMillis(Long.valueOf(fieldValue));
                        record.setField(i, timestampData);
                        break;
                    case BINARY:
                        record.setField(i, Arrays.toString(fieldValue.getBytes(StandardCharsets.UTF_8)));
                        break;

                    // support mysql
                    case INTEGER:
                        record.setField(i, Integer.valueOf(fieldValue));
                        break;
                    case SMALLINT:
                        record.setField(i, Short.valueOf(fieldValue));
                        break;
                    case TINYINT:
                        record.setField(i, Byte.valueOf(fieldValue));
                        break;
                    case FLOAT:
                        record.setField(i, Float.valueOf(fieldValue));
                        break;
                    case DATE:
                        record.setField(i, (int) LocalDate.parse(fieldValue).toEpochDay());
                        break;
                    case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
                        TemporalAccessor parsedTimestampWithLocalZone =
                                SQL_TIMESTAMP_WITH_LOCAL_TIMEZONE_FORMAT.parse(fieldValue);
                        LocalTime localTime = parsedTimestampWithLocalZone.query(TemporalQueries.localTime());
                        LocalDate localDate = parsedTimestampWithLocalZone.query(TemporalQueries.localDate());
                        record.setField(i, TimestampData.fromInstant(LocalDateTime.of(localDate, localTime)
                                .toInstant(ZoneOffset.UTC)));
                        break;
                    case TIMESTAMP_WITHOUT_TIME_ZONE:
                        fieldValue = fieldValue.replace("T", " ");
                        TimestampData timestamp = TimestampData.fromTimestamp(Timestamp.valueOf(fieldValue));
                        record.setField(i, timestamp);
                        break;
                    default:
                        record.setField(i, StringData.fromString(fieldValue));
                }
            }
        }
        return record;
    }
Convert fieldMap(data) to GenericRowData with rowType(schema)
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/jdbc/src/main/java/org/apache/inlong/sort/jdbc/internal/JdbcMultiBatchingOutputFormat.java