[INLONG-11966][Sort]The deserialization process supports returning the data byte size in one rowdata (#11967)

Author: baomingyu

inlong-sort/sort-formats/format-common/src/main/java/org/apache/inlong/sort/formats/base/FormatMsg.java

@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.sort.formats.base;
+
+import lombok.Data;
+import org.apache.flink.table.data.RowData;
+
+@Data
+public class FormatMsg {
+
+    private RowData rowData;
+    private long rowDataLength;
+
+    public FormatMsg(RowData rowData, long rowDataLength) {
+        this.rowData = rowData;
+        this.rowDataLength = rowDataLength;
+    }
+}

inlong-sort/sort-formats/format-common/src/main/java/org/apache/inlong/sort/formats/base/TableFormatUtils.java

@@ -60,6 +60,7 @@
 import org.apache.inlong.common.pojo.sort.dataflow.field.format.VarCharFormatInfo;
 import org.apache.inlong.sort.formats.inlongmsg.FailureHandler;
 
+import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.common.typeinfo.Types;
 import org.apache.flink.table.api.DataTypes;
@@ -582,6 +583,35 @@ public static Object deserializeBasicField(
         return null;
     }
 
+    public static long getFormatValueLength(FormatInfo fieldFormatInfo, String fieldText) {
+        if (fieldFormatInfo instanceof BooleanFormatInfo) {
+            return 4;
+        } else if (fieldFormatInfo instanceof ByteFormatInfo) {
+            return 4;
+        } else if (fieldFormatInfo instanceof BooleanFormatInfo) {
+            return 4;
+        } else if (fieldFormatInfo instanceof ShortFormatInfo) {
+            return 4;
+        } else if (fieldFormatInfo instanceof IntFormatInfo) {
+            return 4;
+        } else if (fieldFormatInfo instanceof LongFormatInfo) {
+            return 8;
+        } else if (fieldFormatInfo instanceof FloatFormatInfo) {
+            return 8;
+        } else if (fieldFormatInfo instanceof DoubleFormatInfo) {
+            return 8;
+        } else if (fieldFormatInfo instanceof DecimalFormatInfo) {
+            return 8;
+        } else if (fieldFormatInfo instanceof DateFormatInfo
+                || fieldFormatInfo instanceof TimeFormatInfo
+                || fieldFormatInfo instanceof TimestampFormatInfo) {
+            return 8;
+        } else if (StringUtils.isNotEmpty(fieldText)) {
+            return fieldText.length();
+        }
+        return 0L;
+    }
+
     /**
      * Serializes the basic field.
      */

inlong-sort/sort-formats/format-rowdata/format-inlongmsg-rowdata-base/src/main/java/org/apache/inlong/sort/formats/inlongmsg/AbstractInLongMsgDeserializationSchema.java

@@ -17,6 +17,7 @@
 
 package org.apache.inlong.sort.formats.inlongmsg;
 
+import org.apache.inlong.sort.formats.base.FormatMsg;
 import org.apache.inlong.sort.formats.metrics.FormatMetricGroup;
 
 import org.apache.flink.api.common.serialization.DeserializationSchema;
@@ -78,6 +79,14 @@ public void deserialize(byte[] message, Collector<RowData> out) {
         }
     }
 
+    public void deserializeFormatMsg(byte[] message, Collector<FormatMsg> out) {
+        try {
+            formatDeserializer.flatFormatMsgMap(message, out);
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
     public List<InLongMsgWrap> preParse(byte[] bytes) throws Exception {
         return formatDeserializer.preParse(bytes);
     }
@@ -88,6 +97,11 @@ public void parse(
         formatDeserializer.parse(inLongMsgWrap, collector);
     }
 
+    public void parseFormatMsg(InLongMsgWrap inLongMsgWrap,
+            Collector<FormatMsg> collector) throws Exception {
+        formatDeserializer.parseFormatMsg(inLongMsgWrap, collector);
+    }
+
     @Override
     public boolean isEndOfStream(RowData rowData) {
         return false;

inlong-sort/sort-formats/format-rowdata/format-inlongmsg-rowdata-base/src/main/java/org/apache/inlong/sort/formats/inlongmsg/AbstractInLongMsgFormatDeserializer.java

@@ -18,6 +18,7 @@
 package org.apache.inlong.sort.formats.inlongmsg;
 
 import org.apache.inlong.common.msg.InLongMsg;
+import org.apache.inlong.sort.formats.base.FormatMsg;
 import org.apache.inlong.sort.formats.metrics.FormatMetricGroup;
 
 import org.apache.commons.lang3.StringUtils;
@@ -32,6 +33,7 @@
 
 import java.io.IOException;
 import java.io.Serializable;
+import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
@@ -45,6 +47,10 @@ public abstract class AbstractInLongMsgFormatDeserializer implements ResultTypeQ
 
     private static final Logger LOG = LoggerFactory.getLogger(AbstractInLongMsgFormatDeserializer.class);
 
+    protected long lastPrintTimestamp = 0L;
+    protected long PRINT_TIMESTAMP_INTERVAL = 60 * 1000L;
+    protected int fieldNameSize = 0;
+
     protected FailureHandler failureHandler;
 
     /**
@@ -76,6 +82,17 @@ public AbstractInLongMsgFormatDeserializer(@Nonnull FailureHandler failureHandle
      */
     protected abstract List<RowData> convertRowDataList(InLongMsgHead head, InLongMsgBody body) throws Exception;
 
+    protected abstract List<FormatMsg> convertFormatMsgList(InLongMsgHead head, InLongMsgBody body) throws Exception;
+
+    protected boolean needPrint() {
+        long now = Instant.now().toEpochMilli();
+        if (now - lastPrintTimestamp > PRINT_TIMESTAMP_INTERVAL) {
+            lastPrintTimestamp = now;
+            return true;
+        }
+        return false;
+    }
+
     public void flatMap(
             byte[] bytes,
             Collector<RowData> collector) throws Exception {
@@ -84,6 +101,14 @@ public void flatMap(
         }
     }
 
+    public void flatFormatMsgMap(
+            byte[] bytes,
+            Collector<FormatMsg> collector) throws Exception {
+        for (InLongMsgWrap inLongMsgWrap : preParse(bytes)) {
+            parseFormatMsg(inLongMsgWrap, collector);
+        }
+    }
+
     public List<InLongMsgWrap> preParse(byte[] bytes) throws Exception {
         final List<InLongMsgWrap> result = new ArrayList<>();
 
@@ -160,6 +185,26 @@ public void parse(InLongMsgWrap inLongMsgWrap, Collector<RowData> collector) thr
         }
     }
 
+    public void parseFormatMsg(InLongMsgWrap inLongMsgWrap, Collector<FormatMsg> collector) throws Exception {
+        InLongMsgHead inLongMsgHead = inLongMsgWrap.getInLongMsgHead();
+
+        for (InLongMsgBody inLongMsgBody : inLongMsgWrap.getInLongMsgBodyList()) {
+            List<FormatMsg> formatMsgList;
+            try {
+                formatMsgList = convertFormatMsgList(inLongMsgHead, inLongMsgBody);
+            } catch (Exception e) {
+                reportDeSerializeErrorMetrics();
+                failureHandler.onConvertingRowFailure(inLongMsgHead, inLongMsgBody, e);
+                continue;
+            }
+            if (formatMsgList != null) {
+                for (FormatMsg formatMsg : formatMsgList) {
+                    collector.collect(formatMsg);
+                }
+            }
+        }
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {

inlong-sort/sort-formats/format-rowdata/format-inlongmsg-rowdata-base/src/main/java/org/apache/inlong/sort/formats/inlongmsg/InLongMsgUtils.java

@@ -19,6 +19,7 @@
 
 import org.apache.inlong.common.pojo.sort.dataflow.field.format.RowFormatInfo;
 import org.apache.inlong.sort.formats.base.FieldToRowDataConverters;
+import org.apache.inlong.sort.formats.base.FormatMsg;
 import org.apache.inlong.sort.formats.base.TableFormatUtils;
 
 import org.apache.flink.api.common.typeinfo.TypeInformation;
@@ -400,6 +401,17 @@ public static GenericRowData decorateRowWithMetaData(
         return producedRow;
     }
 
+    public static FormatMsg decorateFormatMsgWithNeededHeadFields(
+            @Nullable String timeFieldName,
+            @Nullable String attributesFieldName,
+            Timestamp time,
+            Map<String, String> attributes,
+            FormatMsg formatMsg) {
+        formatMsg.setRowData(decorateRowDataWithNeededHeadFields(timeFieldName, attributesFieldName, time, attributes,
+                (GenericRowData) formatMsg.getRowData()));
+        return formatMsg;
+    }
+
     public static GenericRowData decorateRowDataWithNeededHeadFields(
             @Nullable String timeFieldName,
             @Nullable String attributesFieldName,

inlong-sort/sort-formats/format-rowdata/format-inlongmsg-rowdata-binlog/src/main/java/org/apache/inlong/sort/formats/inlongmsgbinlog/InLongMsgBinlogFormatDeserializer.java

@@ -18,6 +18,7 @@
 package org.apache.inlong.sort.formats.inlongmsgbinlog;
 
 import org.apache.inlong.common.pojo.sort.dataflow.field.format.RowFormatInfo;
+import org.apache.inlong.sort.formats.base.FormatMsg;
 import org.apache.inlong.sort.formats.inlongmsg.AbstractInLongMsgFormatDeserializer;
 import org.apache.inlong.sort.formats.inlongmsg.FailureHandler;
 import org.apache.inlong.sort.formats.inlongmsg.InLongMsgBody;
@@ -154,6 +155,19 @@ protected List<RowData> convertRowDataList(InLongMsgHead head, InLongMsgBody bod
                 failureHandler);
     }
 
+    @Override
+    protected List<FormatMsg> convertFormatMsgList(InLongMsgHead head, InLongMsgBody body) throws Exception {
+        return InLongMsgBinlogUtils.getFormatMsgData(
+                rowFormatInfo,
+                timeFieldName,
+                attributesFieldName,
+                metadataFieldName,
+                head.getAttributes(),
+                body.getData(),
+                includeUpdateBefore,
+                failureHandler);
+    }
+
     @Override
     public TypeInformation<RowData> getProducedType() {
         return InLongMsgBinlogUtils.getRowType(