public void open() {
    if (rowDataWrapper != null) {
      return;
    }
    if (rowType != null && struct != null) {
      rowDataWrapper = new RowDataWrapper(rowType, struct);
    }
  }
Should open firstly to initial RowDataWrapper, because it cannot be serialized.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/shuffle/ShuffleHelper.java