public Object convertIfNecessary(Type primitiveType, Object obj) {
          if (obj == null) {
            return null;
          }
          switch (primitiveType.typeId()) {
            case STRING:
              obj = StringData.fromString(obj.toString());
              break;
            case TIME:
              obj = (int) (((long) obj) / 1000_1000);
              break;
            case TIMESTAMP:
              Types.TimestampType timestamp = (Types.TimestampType) primitiveType;
              if (timestamp.shouldAdjustToUTC()) {
                obj = TimestampData.fromInstant((Instant) obj);
              } else {
                obj = TimestampData.fromLocalDateTime((LocalDateTime) obj);
              }
              break;
            case DECIMAL:
              Types.DecimalType decimalType = (Types.DecimalType) primitiveType;
              int precision = decimalType.precision();
              int scale = decimalType.scale();
              obj = DecimalData.fromBigDecimal((BigDecimal) obj, precision, scale);
              break;
            case LIST:
              assert obj instanceof LogFlinkArrayData;
              LogFlinkArrayData list = (LogFlinkArrayData) obj;
              obj = list.arrayData;
              break;
            case MAP:
              assert obj instanceof LogFlinkMapData;
              LogFlinkMapData map = (LogFlinkMapData) obj;
              obj = map.mapData;
              break;
          }
          return obj;
        }
An implement of {@link LogData<RowData>} in flink application.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common-format/src/main/java/org/apache/amoro/flink/shuffle/LogRecordV1.java