private static RowFieldConverter createRowFieldConverter(LogicalType fieldType) {
        switch (fieldType.getTypeRoot()) {
            case NULL:
                return (row, pos) -> null;
            case BOOLEAN:
                return RowData::getBoolean;
            case TINYINT:
                return RowData::getByte;
            case SMALLINT:
                return RowData::getShort;
            case INTEGER:
            case INTERVAL_YEAR_MONTH:
                return RowData::getInt;
            case BIGINT:
            case INTERVAL_DAY_TIME:
                return RowData::getLong;
            case FLOAT:
                return RowData::getFloat;
            case DOUBLE:
                return RowData::getDouble;
            case CHAR:
            case VARCHAR:
                return (row, pos) -> row.getString(pos).toString();
            case BINARY:
            case VARBINARY:
                return RowData::getBinary;
            case DATE:
                return (row, pos) -> convertDate(row.getLong(pos));
            case TIME_WITHOUT_TIME_ZONE:
                return (row, pos) -> convertTime(row.getInt(pos));
            case TIMESTAMP_WITHOUT_TIME_ZONE:
                final int timestampPrecision = ((TimestampType) fieldType).getPrecision();
                return (row, pos) -> convertTimestamp(
                        row.getTimestamp(pos, timestampPrecision));
            case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
                final int zonedTimestampPrecision =
                        ((LocalZonedTimestampType) fieldType).getPrecision();
                return (row, pos) -> convertTimestamp(
                        row.getTimestamp(pos, zonedTimestampPrecision));
            case DECIMAL:
                return createDecimalRowFieldConverter((DecimalType) fieldType);
            case ARRAY:
                return createArrayRowFieldConverter((ArrayType) fieldType);
            case ROW:
                return createRowRowFieldConverter((RowType) fieldType);
            case MAP:
            case MULTISET:
            case RAW:
            default:
                throw new UnsupportedOperationException("Unsupported type: " + fieldType);
        }
    }
Runtime converter that converts objects of Flink Table & SQL internal data structures to
corresponding {@link Object}s.
inlong-sort/sort-formats/format-rowdata/format-rowdata-base/src/main/java/org/apache/inlong/sort/formats/base/RowDataToFieldConverters.java