[INLONG-10784][Sort] Refactor the structure of inlongmsg-rowdata

apache · Aug 13, 2024 · a69075d · a69075d
1 parent ef20f56
commit a69075d
Show file tree

Hide file tree

Showing 32 changed files with 117 additions and 102 deletions.
diff --git a/...gmsg/AbstractInLongMsgDecodingFormat.java → ...data/AbstractInLongMsgDecodingFormat.java b/...gmsg/AbstractInLongMsgDecodingFormat.java → ...data/AbstractInLongMsgDecodingFormat.java
@@ -15,7 +15,9 @@
  * limitations under the License.
  */
 
-package org.apache.inlong.sort.formats.inlongmsg;
+package org.apache.inlong.sort.formats.inlongmsg.rowdata;
+
+import org.apache.inlong.sort.formats.inlongmsg.InLongMsgMetadata;
 
 import org.apache.flink.api.common.serialization.DeserializationSchema;
 import org.apache.flink.table.connector.format.DecodingFormat;

diff --git a/...stractInLongMsgDeserializationSchema.java → ...stractInLongMsgDeserializationSchema.java b/...stractInLongMsgDeserializationSchema.java → ...stractInLongMsgDeserializationSchema.java
@@ -15,8 +15,9 @@
  * limitations under the License.
  */
 
-package org.apache.inlong.sort.formats.inlongmsg;
+package org.apache.inlong.sort.formats.inlongmsg.rowdata;
 
+import org.apache.inlong.sort.formats.inlongmsg.InLongMsgWrap;
 import org.apache.inlong.sort.formats.metrics.FormatMetricGroup;
 
 import org.apache.flink.api.common.serialization.DeserializationSchema;

diff --git a/.../AbstractInLongMsgFormatDeserializer.java → .../AbstractInLongMsgFormatDeserializer.java b/.../AbstractInLongMsgFormatDeserializer.java → .../AbstractInLongMsgFormatDeserializer.java
@@ -15,9 +15,14 @@
  * limitations under the License.
  */
 
-package org.apache.inlong.sort.formats.inlongmsg;
+package org.apache.inlong.sort.formats.inlongmsg.rowdata;
 
 import org.apache.inlong.common.msg.InLongMsg;
+import org.apache.inlong.sort.formats.inlongmsg.FailureHandler;
+import org.apache.inlong.sort.formats.inlongmsg.IgnoreFailureHandler;
+import org.apache.inlong.sort.formats.inlongmsg.InLongMsgBody;
+import org.apache.inlong.sort.formats.inlongmsg.InLongMsgHead;
+import org.apache.inlong.sort.formats.inlongmsg.InLongMsgWrap;
 import org.apache.inlong.sort.formats.metrics.FormatMetricGroup;
 
 import org.apache.flink.api.java.typeutils.ResultTypeQueryable;

diff --git a/...bstractInLongMsgMixedFormatConverter.java → ...bstractInLongMsgMixedFormatConverter.java b/...bstractInLongMsgMixedFormatConverter.java → ...bstractInLongMsgMixedFormatConverter.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.inlong.sort.formats.inlongmsg;
+package org.apache.inlong.sort.formats.inlongmsg.rowdata;
 
 import org.apache.flink.metrics.MetricGroup;
 import org.apache.flink.table.data.RowData;

diff --git a/...ractInLongMsgMixedFormatDeserializer.java → ...ractInLongMsgMixedFormatDeserializer.java b/...ractInLongMsgMixedFormatDeserializer.java → ...ractInLongMsgMixedFormatDeserializer.java
@@ -15,7 +15,9 @@
  * limitations under the License.
  */
 
-package org.apache.inlong.sort.formats.inlongmsg;
+package org.apache.inlong.sort.formats.inlongmsg.rowdata;
+
+import org.apache.inlong.sort.formats.inlongmsg.FailureHandler;
 
 import javax.annotation.Nonnull;
 

diff --git a/...ongmsg/InLongMsgMixedFormatConverter.java → ...owdata/InLongMsgMixedFormatConverter.java b/...ongmsg/InLongMsgMixedFormatConverter.java → ...owdata/InLongMsgMixedFormatConverter.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.inlong.sort.formats.inlongmsg;
+package org.apache.inlong.sort.formats.inlongmsg.rowdata;
 
 import org.apache.flink.api.common.functions.FlatMapFunction;
 import org.apache.flink.api.java.typeutils.ResultTypeQueryable;

diff --git a/...InLongMsgMixedFormatConverterBuilder.java → ...InLongMsgMixedFormatConverterBuilder.java b/...InLongMsgMixedFormatConverterBuilder.java → ...InLongMsgMixedFormatConverterBuilder.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.inlong.sort.formats.inlongmsg;
+package org.apache.inlong.sort.formats.inlongmsg.rowdata;
 
 import org.apache.inlong.common.pojo.sort.dataflow.field.format.RowFormatInfo;
 
@@ -27,8 +27,8 @@
 import static org.apache.inlong.sort.formats.base.TableFormatConstants.FORMAT_IGNORE_ERRORS;
 import static org.apache.inlong.sort.formats.base.TableFormatConstants.FORMAT_NULL_LITERAL;
 import static org.apache.inlong.sort.formats.base.TableFormatConstants.FORMAT_TIME_FIELD_NAME;
-import static org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.DEFAULT_ATTRIBUTES_FIELD_NAME;
-import static org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.DEFAULT_TIME_FIELD_NAME;
+import static org.apache.inlong.sort.formats.inlongmsg.rowdata.InLongMsgUtils.DEFAULT_ATTRIBUTES_FIELD_NAME;
+import static org.apache.inlong.sort.formats.inlongmsg.rowdata.InLongMsgUtils.DEFAULT_TIME_FIELD_NAME;
 
 /**
  * The builder for {@link AbstractInLongMsgMixedFormatConverter}s.

diff --git a/...LongMsgMixedFormatConverterValidator.java → ...LongMsgMixedFormatConverterValidator.java b/...LongMsgMixedFormatConverterValidator.java → ...LongMsgMixedFormatConverterValidator.java
@@ -15,15 +15,15 @@
  * limitations under the License.
  */
 
-package org.apache.inlong.sort.formats.inlongmsg;
+package org.apache.inlong.sort.formats.inlongmsg.rowdata;
 
 import org.apache.inlong.sort.formats.base.FormatDescriptorValidator;
 
 import org.apache.flink.table.descriptors.DescriptorProperties;
 
 import static org.apache.inlong.sort.formats.base.TableFormatConstants.FORMAT_IGNORE_ERRORS;
 import static org.apache.inlong.sort.formats.base.TableFormatConstants.FORMAT_NULL_LITERAL;
-import static org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.validateInLongMsgSchema;
+import static org.apache.inlong.sort.formats.inlongmsg.rowdata.InLongMsgUtils.validateInLongMsgSchema;
 
 /**
  * The validator for the properties of {@link AbstractInLongMsgMixedFormatConverter}s.

diff --git a/...gMsgMixedFormatDeserializerValidator.java → ...gMsgMixedFormatDeserializerValidator.java b/...gMsgMixedFormatDeserializerValidator.java → ...gMsgMixedFormatDeserializerValidator.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.inlong.sort.formats.inlongmsg;
+package org.apache.inlong.sort.formats.inlongmsg.rowdata;
 
 import org.apache.inlong.sort.formats.base.FormatDescriptorValidator;
 

diff --git a/...sgTextMixedFormatDeserializerBuilder.java → ...sgTextMixedFormatDeserializerBuilder.java b/...sgTextMixedFormatDeserializerBuilder.java → ...sgTextMixedFormatDeserializerBuilder.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.inlong.sort.formats.inlongmsg;
+package org.apache.inlong.sort.formats.inlongmsg.rowdata;
 
 import org.apache.flink.table.descriptors.DescriptorProperties;
 

diff --git a/...ort/formats/inlongmsg/InLongMsgUtils.java → ...ats/inlongmsg/rowdata/InLongMsgUtils.java b/...ort/formats/inlongmsg/InLongMsgUtils.java → ...ats/inlongmsg/rowdata/InLongMsgUtils.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.inlong.sort.formats.inlongmsg;
+package org.apache.inlong.sort.formats.inlongmsg.rowdata;
 
 import org.apache.inlong.common.pojo.sort.dataflow.field.format.RowFormatInfo;
 import org.apache.inlong.sort.formats.base.FieldToRowDataConverters;
@@ -33,6 +33,11 @@
 import org.apache.flink.table.types.logical.RowType;
 import org.apache.flink.table.types.logical.TimestampType;
 import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.inlong.sort.formats.inlongmsg.IgnoreFailureHandler;
+import org.apache.inlong.sort.formats.inlongmsg.InLongMsgHead;
+import org.apache.inlong.sort.formats.inlongmsg.InLongMsgMetadata;
+import org.apache.inlong.sort.formats.inlongmsg.FailureHandler;
+import org.apache.inlong.sort.formats.inlongmsg.NoOpFailureHandler;
 
 import javax.annotation.Nullable;
 
@@ -212,15 +217,15 @@ public static TypeInformation<RowData> decorateRowTypeWithMetadata(
         // Metadata
         final List<InLongMsgMetadata.ReadableMetadata> readableMetadata = metadataKeys.stream()
                 .map(k -> Stream.of(InLongMsgMetadata.ReadableMetadata.values())
-                        .filter(rm -> rm.key.equals(k))
+                        .filter(rm -> rm.getKey().equals(k))
                         .findFirst()
                         .orElseThrow(IllegalStateException::new))
                 .collect(Collectors.toList());
 
         for (InLongMsgMetadata.ReadableMetadata rm : readableMetadata) {
             // TODO : avoid name conflict?
-            fieldNames.add(rm.key);
-            fieldTypes.add(rm.dataType.getLogicalType());
+            fieldNames.add(rm.getKey());
+            fieldTypes.add(rm.getDataType().getLogicalType());
         }
 
         RowType rowType = RowType.of(fieldTypes.toArray(new LogicalType[0]), fieldNames.toArray(new String[0]));
@@ -319,15 +324,15 @@ public static TypeInformation<RowData> decorateRowTypeWithNeededHeadFieldsAndMet
         // Metadata
         final List<InLongMsgMetadata.ReadableMetadata> readableMetadata = metadataKeys.stream()
                 .map(k -> Stream.of(InLongMsgMetadata.ReadableMetadata.values())
-                        .filter(rm -> rm.key.equals(k))
+                        .filter(rm -> rm.getKey().equals(k))
                         .findFirst()
                         .orElseThrow(IllegalStateException::new))
                 .collect(Collectors.toList());
 
         for (InLongMsgMetadata.ReadableMetadata rm : readableMetadata) {
             // TODO : avoid name conflict?
-            fieldNames.add(rm.key);
-            fieldTypes.add(rm.dataType.getLogicalType());
+            fieldNames.add(rm.getKey());
+            fieldTypes.add(rm.getDataType().getLogicalType());
         }
 
         RowType rowType = RowType.of(fieldTypes.toArray(new LogicalType[0]), fieldNames.toArray(new String[0]));

diff --git a/...in/java/org/apache/inlong/sort/formats/inlongmsgbinlog/InLongMsgBinlogDecodingFormat.java b/...in/java/org/apache/inlong/sort/formats/inlongmsgbinlog/InLongMsgBinlogDecodingFormat.java
@@ -17,7 +17,7 @@
 
 package org.apache.inlong.sort.formats.inlongmsgbinlog;
 
-import org.apache.inlong.sort.formats.inlongmsg.AbstractInLongMsgDecodingFormat;
+import org.apache.inlong.sort.formats.inlongmsg.rowdata.AbstractInLongMsgDecodingFormat;
 
 import org.apache.flink.api.common.serialization.DeserializationSchema;
 import org.apache.flink.configuration.ReadableConfig;

diff --git a/...ava/org/apache/inlong/sort/formats/inlongmsgbinlog/InLongMsgBinlogFormatDeserializer.java b/...ava/org/apache/inlong/sort/formats/inlongmsgbinlog/InLongMsgBinlogFormatDeserializer.java
@@ -18,11 +18,11 @@
 package org.apache.inlong.sort.formats.inlongmsgbinlog;
 
 import org.apache.inlong.common.pojo.sort.dataflow.field.format.RowFormatInfo;
-import org.apache.inlong.sort.formats.inlongmsg.AbstractInLongMsgFormatDeserializer;
+import org.apache.inlong.sort.formats.inlongmsg.rowdata.AbstractInLongMsgFormatDeserializer;
 import org.apache.inlong.sort.formats.inlongmsg.FailureHandler;
 import org.apache.inlong.sort.formats.inlongmsg.InLongMsgBody;
 import org.apache.inlong.sort.formats.inlongmsg.InLongMsgHead;
-import org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils;
+import org.apache.inlong.sort.formats.inlongmsg.rowdata.InLongMsgUtils;
 
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.table.data.RowData;

diff --git a/...ache/inlong/sort/formats/inlongmsgbinlog/InLongMsgBinlogRowDataDeserializationSchema.java b/...ache/inlong/sort/formats/inlongmsgbinlog/InLongMsgBinlogRowDataDeserializationSchema.java
@@ -18,15 +18,15 @@
 package org.apache.inlong.sort.formats.inlongmsgbinlog;
 
 import org.apache.inlong.common.pojo.sort.dataflow.field.format.RowFormatInfo;
-import org.apache.inlong.sort.formats.inlongmsg.AbstractInLongMsgDeserializationSchema;
-import org.apache.inlong.sort.formats.inlongmsg.AbstractInLongMsgFormatDeserializer;
+import org.apache.inlong.sort.formats.inlongmsg.rowdata.AbstractInLongMsgDeserializationSchema;
+import org.apache.inlong.sort.formats.inlongmsg.rowdata.AbstractInLongMsgFormatDeserializer;
 
 import org.apache.flink.annotation.PublicEvolving;
 
 import static org.apache.inlong.sort.formats.base.TableFormatConstants.DEFAULT_INCLUDE_UPDATE_BEFORE;
 import static org.apache.inlong.sort.formats.base.TableFormatConstants.DEFAULT_METADATA_FIELD_NAME;
-import static org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.DEFAULT_ATTRIBUTES_FIELD_NAME;
-import static org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.DEFAULT_TIME_FIELD_NAME;
+import static org.apache.inlong.sort.formats.inlongmsg.rowdata.InLongMsgUtils.DEFAULT_ATTRIBUTES_FIELD_NAME;
+import static org.apache.inlong.sort.formats.inlongmsg.rowdata.InLongMsgUtils.DEFAULT_TIME_FIELD_NAME;
 
 /**
  * Deserialization schema from inlong-msg-binlog to Flink Table & SQL internal data structures.

diff --git a/...og/src/main/java/org/apache/inlong/sort/formats/inlongmsgbinlog/InLongMsgBinlogUtils.java b/...og/src/main/java/org/apache/inlong/sort/formats/inlongmsgbinlog/InLongMsgBinlogUtils.java
@@ -43,13 +43,13 @@
 import java.util.Map;
 
 import static org.apache.inlong.sort.formats.base.TableFormatUtils.deserializeBasicField;
-import static org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_INTERFACE_ID;
-import static org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_INTERFACE_NAME;
-import static org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_INTERFACE_TID;
-import static org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_STREAM_ID;
-import static org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.getPredefinedFields;
-import static org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.parseAttr;
-import static org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.parseEpochTime;
+import static org.apache.inlong.sort.formats.inlongmsg.rowdata.InLongMsgUtils.INLONGMSG_ATTR_INTERFACE_ID;
+import static org.apache.inlong.sort.formats.inlongmsg.rowdata.InLongMsgUtils.INLONGMSG_ATTR_INTERFACE_NAME;
+import static org.apache.inlong.sort.formats.inlongmsg.rowdata.InLongMsgUtils.INLONGMSG_ATTR_INTERFACE_TID;
+import static org.apache.inlong.sort.formats.inlongmsg.rowdata.InLongMsgUtils.INLONGMSG_ATTR_STREAM_ID;
+import static org.apache.inlong.sort.formats.inlongmsg.rowdata.InLongMsgUtils.getPredefinedFields;
+import static org.apache.inlong.sort.formats.inlongmsg.rowdata.InLongMsgUtils.parseAttr;
+import static org.apache.inlong.sort.formats.inlongmsg.rowdata.InLongMsgUtils.parseEpochTime;
 
 /**
  * Utilities for InLongMsgBinlog.

diff --git a/...src/main/java/org/apache/inlong/sort/formats/inlongmsgcsv/InLongMsgCsvDecodingFormat.java b/...src/main/java/org/apache/inlong/sort/formats/inlongmsgcsv/InLongMsgCsvDecodingFormat.java
@@ -17,7 +17,7 @@
 
 package org.apache.inlong.sort.formats.inlongmsgcsv;
 
-import org.apache.inlong.sort.formats.inlongmsg.AbstractInLongMsgDecodingFormat;
+import org.apache.inlong.sort.formats.inlongmsg.rowdata.AbstractInLongMsgDecodingFormat;
 
 import org.apache.commons.lang3.StringEscapeUtils;
 import org.apache.flink.api.common.serialization.DeserializationSchema;

diff --git a/...main/java/org/apache/inlong/sort/formats/inlongmsgcsv/InLongMsgCsvFormatDeserializer.java b/...main/java/org/apache/inlong/sort/formats/inlongmsgcsv/InLongMsgCsvFormatDeserializer.java
@@ -22,11 +22,11 @@
 import org.apache.inlong.sort.formats.base.FieldToRowDataConverters.FieldToRowDataConverter;
 import org.apache.inlong.sort.formats.base.TableFormatUtils;
 import org.apache.inlong.sort.formats.base.TextFormatBuilder;
-import org.apache.inlong.sort.formats.inlongmsg.AbstractInLongMsgFormatDeserializer;
+import org.apache.inlong.sort.formats.inlongmsg.rowdata.AbstractInLongMsgFormatDeserializer;
 import org.apache.inlong.sort.formats.inlongmsg.FailureHandler;
 import org.apache.inlong.sort.formats.inlongmsg.InLongMsgBody;
 import org.apache.inlong.sort.formats.inlongmsg.InLongMsgHead;
-import org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils;
+import org.apache.inlong.sort.formats.inlongmsg.rowdata.InLongMsgUtils;
 
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.table.data.GenericRowData;
@@ -43,8 +43,8 @@
 import static org.apache.inlong.sort.formats.base.TableFormatConstants.DEFAULT_DELIMITER;
 import static org.apache.inlong.sort.formats.base.TableFormatConstants.DEFAULT_LINE_DELIMITER;
 import static org.apache.inlong.sort.formats.base.TableFormatConstants.DEFAULT_RETAIN_PREDEFINED_FIELD;
-import static org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.DEFAULT_ATTRIBUTES_FIELD_NAME;
-import static org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.DEFAULT_TIME_FIELD_NAME;
+import static org.apache.inlong.sort.formats.inlongmsg.rowdata.InLongMsgUtils.DEFAULT_ATTRIBUTES_FIELD_NAME;
+import static org.apache.inlong.sort.formats.inlongmsg.rowdata.InLongMsgUtils.DEFAULT_TIME_FIELD_NAME;
 import static org.apache.inlong.sort.formats.inlongmsgcsv.InLongMsgCsvUtils.DEFAULT_DELETE_HEAD_DELIMITER;
 
 /**

diff --git a/...org/apache/inlong/sort/formats/inlongmsgcsv/InLongMsgCsvRowDataDeserializationSchema.java b/...org/apache/inlong/sort/formats/inlongmsgcsv/InLongMsgCsvRowDataDeserializationSchema.java
@@ -18,8 +18,8 @@
 package org.apache.inlong.sort.formats.inlongmsgcsv;
 
 import org.apache.inlong.common.pojo.sort.dataflow.field.format.RowFormatInfo;
-import org.apache.inlong.sort.formats.inlongmsg.AbstractInLongMsgDeserializationSchema;
-import org.apache.inlong.sort.formats.inlongmsg.AbstractInLongMsgFormatDeserializer;
+import org.apache.inlong.sort.formats.inlongmsg.rowdata.AbstractInLongMsgDeserializationSchema;
+import org.apache.inlong.sort.formats.inlongmsg.rowdata.AbstractInLongMsgFormatDeserializer;
 
 import org.apache.flink.annotation.PublicEvolving;
 
@@ -32,8 +32,8 @@
 import static org.apache.inlong.sort.formats.base.TableFormatConstants.DEFAULT_LINE_DELIMITER;
 import static org.apache.inlong.sort.formats.base.TableFormatConstants.DEFAULT_NULL_LITERAL;
 import static org.apache.inlong.sort.formats.base.TableFormatConstants.DEFAULT_QUOTE_CHARACTER;
-import static org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.DEFAULT_ATTRIBUTES_FIELD_NAME;
-import static org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.DEFAULT_TIME_FIELD_NAME;
+import static org.apache.inlong.sort.formats.inlongmsg.rowdata.InLongMsgUtils.DEFAULT_ATTRIBUTES_FIELD_NAME;
+import static org.apache.inlong.sort.formats.inlongmsg.rowdata.InLongMsgUtils.DEFAULT_TIME_FIELD_NAME;
 
 /**
  * Deserialization schema from InLongMsg-CSV to Flink Table & SQL internal data structures.

diff --git a/...data-csv/src/main/java/org/apache/inlong/sort/formats/inlongmsgcsv/InLongMsgCsvUtils.java b/...data-csv/src/main/java/org/apache/inlong/sort/formats/inlongmsgcsv/InLongMsgCsvUtils.java
@@ -37,16 +37,16 @@
 import java.util.stream.Collectors;
 
 import static org.apache.inlong.sort.formats.base.TableFormatUtils.deserializeBasicField;
-import static org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_INTERFACE_ID;
-import static org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_INTERFACE_NAME;
-import static org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_INTERFACE_TID;
-import static org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_STREAM_ID;
-import static org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_TIME_DT;
-import static org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_TIME_T;
-import static org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.getPredefinedFields;
-import static org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.parseAttr;
-import static org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.parseDateTime;
-import static org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.parseEpochTime;
+import static org.apache.inlong.sort.formats.inlongmsg.rowdata.InLongMsgUtils.INLONGMSG_ATTR_INTERFACE_ID;
+import static org.apache.inlong.sort.formats.inlongmsg.rowdata.InLongMsgUtils.INLONGMSG_ATTR_INTERFACE_NAME;
+import static org.apache.inlong.sort.formats.inlongmsg.rowdata.InLongMsgUtils.INLONGMSG_ATTR_INTERFACE_TID;
+import static org.apache.inlong.sort.formats.inlongmsg.rowdata.InLongMsgUtils.INLONGMSG_ATTR_STREAM_ID;
+import static org.apache.inlong.sort.formats.inlongmsg.rowdata.InLongMsgUtils.INLONGMSG_ATTR_TIME_DT;
+import static org.apache.inlong.sort.formats.inlongmsg.rowdata.InLongMsgUtils.INLONGMSG_ATTR_TIME_T;
+import static org.apache.inlong.sort.formats.inlongmsg.rowdata.InLongMsgUtils.getPredefinedFields;
+import static org.apache.inlong.sort.formats.inlongmsg.rowdata.InLongMsgUtils.parseAttr;
+import static org.apache.inlong.sort.formats.inlongmsg.rowdata.InLongMsgUtils.parseDateTime;
+import static org.apache.inlong.sort.formats.inlongmsg.rowdata.InLongMsgUtils.parseEpochTime;
 import static org.apache.inlong.sort.formats.util.StringUtils.splitCsv;
 
 /**

diff --git a/.../java/org/apache/inlong/sort/formats/inlongmsgcsv/InLongMsgCsvFormatDeserializerTest.java b/.../java/org/apache/inlong/sort/formats/inlongmsgcsv/InLongMsgCsvFormatDeserializerTest.java
@@ -59,8 +59,8 @@
 import static org.apache.inlong.sort.formats.base.TableFormatConstants.DEFAULT_CHARSET;
 import static org.apache.inlong.sort.formats.base.TableFormatConstants.DEFAULT_DELIMITER;
 import static org.apache.inlong.sort.formats.inlongmsg.InLongMsgMetadata.ReadableMetadata.STREAMID;
-import static org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.DEFAULT_ATTRIBUTES_FIELD_NAME;
-import static org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.DEFAULT_TIME_FIELD_NAME;
+import static org.apache.inlong.sort.formats.inlongmsg.rowdata.InLongMsgUtils.DEFAULT_ATTRIBUTES_FIELD_NAME;
+import static org.apache.inlong.sort.formats.inlongmsg.rowdata.InLongMsgUtils.DEFAULT_TIME_FIELD_NAME;
 import static org.junit.Assert.assertEquals;
 
 /**

diff --git a/...v/src/main/java/org/apache/inlong/sort/formats/inlongmsgkv/InLongMsgKvDecodingFormat.java b/...v/src/main/java/org/apache/inlong/sort/formats/inlongmsgkv/InLongMsgKvDecodingFormat.java
@@ -17,7 +17,7 @@
 
 package org.apache.inlong.sort.formats.inlongmsgkv;
 
-import org.apache.inlong.sort.formats.inlongmsg.AbstractInLongMsgDecodingFormat;
+import org.apache.inlong.sort.formats.inlongmsg.rowdata.AbstractInLongMsgDecodingFormat;
 
 import org.apache.commons.lang3.StringEscapeUtils;
 import org.apache.flink.api.common.serialization.DeserializationSchema;

diff --git a/...c/main/java/org/apache/inlong/sort/formats/inlongmsgkv/InLongMsgKvFormatDeserializer.java b/...c/main/java/org/apache/inlong/sort/formats/inlongmsgkv/InLongMsgKvFormatDeserializer.java
@@ -22,11 +22,11 @@
 import org.apache.inlong.sort.formats.base.FieldToRowDataConverters.FieldToRowDataConverter;
 import org.apache.inlong.sort.formats.base.TableFormatUtils;
 import org.apache.inlong.sort.formats.base.TextFormatBuilder;
-import org.apache.inlong.sort.formats.inlongmsg.AbstractInLongMsgFormatDeserializer;
+import org.apache.inlong.sort.formats.inlongmsg.rowdata.AbstractInLongMsgFormatDeserializer;
 import org.apache.inlong.sort.formats.inlongmsg.FailureHandler;
 import org.apache.inlong.sort.formats.inlongmsg.InLongMsgBody;
 import org.apache.inlong.sort.formats.inlongmsg.InLongMsgHead;
-import org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils;
+import org.apache.inlong.sort.formats.inlongmsg.rowdata.InLongMsgUtils;
 
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.table.data.GenericRowData;
@@ -43,8 +43,8 @@
 import static org.apache.inlong.sort.formats.base.TableFormatConstants.DEFAULT_ENTRY_DELIMITER;
 import static org.apache.inlong.sort.formats.base.TableFormatConstants.DEFAULT_KV_DELIMITER;
 import static org.apache.inlong.sort.formats.base.TableFormatConstants.DEFAULT_LINE_DELIMITER;
-import static org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.DEFAULT_ATTRIBUTES_FIELD_NAME;
-import static org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.DEFAULT_TIME_FIELD_NAME;
+import static org.apache.inlong.sort.formats.inlongmsg.rowdata.InLongMsgUtils.DEFAULT_ATTRIBUTES_FIELD_NAME;
+import static org.apache.inlong.sort.formats.inlongmsg.rowdata.InLongMsgUtils.DEFAULT_TIME_FIELD_NAME;
 
 /**
  * The deserializer for the records in InLongMsgKv format.