public static SeaTunnelRow convert(
            InternalRow rowData, SeaTunnelRowType seaTunnelRowType, TableSchema tableSchema) {
        Object[] objects = new Object[seaTunnelRowType.getTotalFields()];
        for (int i = 0; i < objects.length; i++) {
            // judge the field is or not equals null
            if (rowData.isNullAt(i)) {
                objects[i] = null;
                continue;
            }
            SeaTunnelDataType<?> fieldType = seaTunnelRowType.getFieldType(i);
            String fieldName = seaTunnelRowType.getFieldName(i);
            switch (fieldType.getSqlType()) {
                case TINYINT:
                    objects[i] = rowData.getByte(i);
                    break;
                case SMALLINT:
                    objects[i] = rowData.getShort(i);
                    break;
                case INT:
                    objects[i] = rowData.getInt(i);
                    break;
                case BIGINT:
                    objects[i] = rowData.getLong(i);
                    break;
                case FLOAT:
                    objects[i] = rowData.getFloat(i);
                    break;
                case DOUBLE:
                    objects[i] = rowData.getDouble(i);
                    break;
                case DECIMAL:
                    Decimal decimal =
                            rowData.getDecimal(
                                    i,
                                    ((DecimalType) fieldType).getPrecision(),
                                    ((DecimalType) fieldType).getScale());
                    objects[i] = decimal.toBigDecimal();
                    break;
                case STRING:
                    objects[i] = rowData.getString(i).toString();
                    break;
                case BOOLEAN:
                    objects[i] = rowData.getBoolean(i);
                    break;
                case BYTES:
                    objects[i] = rowData.getBinary(i);
                    break;
                case DATE:
                    int dateInt = rowData.getInt(i);
                    objects[i] = DateTimeUtils.toLocalDate(dateInt);
                    break;
                case TIMESTAMP:
                    int precision = TimestampType.DEFAULT_PRECISION;
                    Optional<DataField> precisionOptional =
                            tableSchema.fields().stream()
                                    .filter(dataField -> dataField.name().equals(fieldName))
                                    .findFirst();
                    if (precisionOptional.isPresent()) {
                        precision = ((TimestampType) precisionOptional.get().type()).getPrecision();
                    }
                    Timestamp timestamp = rowData.getTimestamp(i, precision);
                    objects[i] = timestamp.toLocalDateTime();
                    break;
                case ARRAY:
                    InternalArray paimonArray = rowData.getArray(i);
                    ArrayType<?, ?> seatunnelArray = (ArrayType<?, ?>) fieldType;
                    objects[i] =
                            convertArrayType(
                                    fieldName, paimonArray, seatunnelArray.getElementType());
                    break;
                case MAP:
                    MapType<?, ?> mapType = (MapType<?, ?>) fieldType;
                    InternalMap map = rowData.getMap(i);
                    InternalArray keyArray = map.keyArray();
                    InternalArray valueArray = map.valueArray();
                    SeaTunnelDataType<?> keyType = mapType.getKeyType();
                    SeaTunnelDataType<?> valueType = mapType.getValueType();
                    Object[] key = (Object[]) convertArrayType(fieldName, keyArray, keyType);
                    Object[] value = (Object[]) convertArrayType(fieldName, valueArray, valueType);
                    Map<Object, Object> mapData = new HashMap<>();
                    for (int j = 0; j < key.length; j++) {
                        mapData.put(key[j], value[j]);
                    }
                    objects[i] = mapData;
                    break;
                case ROW:
                    SeaTunnelDataType<?> rowType = seaTunnelRowType.getFieldType(i);
                    InternalRow row =
                            rowData.getRow(i, ((SeaTunnelRowType) rowType).getTotalFields());
                    objects[i] = convert(row, (SeaTunnelRowType) rowType, tableSchema);
                    break;
                default:
                    throw CommonError.unsupportedDataType(
                            PaimonConfig.CONNECTOR_IDENTITY,
                            fieldType.getSqlType().toString(),
                            fieldName);
            }
        }
        return new SeaTunnelRow(objects);
    }
@Test
    public void paimonToSeaTunnel() {
        SeaTunnelRow convert = RowConverter.convert(internalRow, seaTunnelRowType, tableSchema);
        Assertions.assertEquals(convert, seaTunnelRow);
    }