[INLONG-11611][SDK] Transform SDK supports RowData source and sink

inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/AvroSourceDecoder.java

@@ -66,7 +66,7 @@ public AvroSourceDecoder(AvroSourceInfo sourceInfo) {
     }
 
     @Override
-    public SourceData decodeBytes(byte[] srcBytes, Context context) {
+    public SourceData decode(byte[] srcBytes, Context context) {
         try {
             InputStream inputStream = new ByteArrayInputStream(srcBytes);
             DataFileStream<GenericRecord> dataFileStream =

inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/BsonSourceDecoder.java

@@ -41,7 +41,7 @@ public BsonSourceDecoder(BsonSourceInfo sourceInfo) {
     }
 
     @Override
-    public SourceData decodeBytes(byte[] srcBytes, Context context) {
+    public SourceData decode(byte[] srcBytes, Context context) {
         return decoder.decode(parse(srcBytes), context);
     }
 

inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/CsvSourceDecoder.java

@@ -52,7 +52,7 @@ public CsvSourceDecoder(CsvSourceInfo sourceInfo) {
     }
 
     @Override
-    public SourceData decodeBytes(byte[] srcBytes, Context context) {
+    public SourceData decode(byte[] srcBytes, Context context) {
         String srcString = new String(srcBytes, srcCharset);
         return this.decode(srcString, context);
     }

inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/JsonSourceDecoder.java

@@ -70,7 +70,7 @@ public JsonSourceDecoder(JsonSourceInfo sourceInfo) {
      * @return
      */
     @Override
-    public SourceData decodeBytes(byte[] srcBytes, Context context) {
+    public SourceData decode(byte[] srcBytes, Context context) {
         String srcString = new String(srcBytes, srcCharset);
         return this.decode(srcString, context);
     }

inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/KvSourceDecoder.java

@@ -66,7 +66,7 @@ public KvSourceDecoder(KvSourceInfo sourceInfo) {
     }
 
     @Override
-    public SourceData decodeBytes(byte[] srcBytes, Context context) {
+    public SourceData decode(byte[] srcBytes, Context context) {
         String srcString = new String(srcBytes, srcCharset);
         return this.decode(srcString, context);
     }

inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/ParquetSourceDecoder.java

@@ -68,7 +68,7 @@ public ParquetSourceDecoder(ParquetSourceInfo sourceInfo) {
 
     @SuppressWarnings("unchecked")
     @Override
-    public SourceData decodeBytes(byte[] srcBytes, Context context) {
+    public SourceData decode(byte[] srcBytes, Context context) {
         try {
             // Create a custom InputFile
             InputFile inputFile = new ParquetInputByteArray(srcBytes);

inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/PbSourceDecoder.java

@@ -96,7 +96,7 @@ public PbSourceDecoder(PbSourceInfo sourceInfo) {
      */
     @SuppressWarnings("unchecked")
     @Override
-    public SourceData decodeBytes(byte[] srcBytes, Context context) {
+    public SourceData decode(byte[] srcBytes, Context context) {
         try {
             // decode
             DynamicMessage.Builder builder = DynamicMessage.newBuilder(rootDesc);
@@ -151,9 +151,4 @@ public SourceData decodeBytes(byte[] srcBytes, Context context) {
             return null;
         }
     }
-
-    @Override
-    public SourceData decode(byte[] bytes, Context context) {
-        return decodeBytes(bytes, context);
-    }
 }

inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/RowDataSourceDecoder.java

@@ -54,7 +54,7 @@ private Map<String, Integer> parseFieldPositionMap(List<FieldInfo> fields) {
     }
 
     @Override
-    public SourceData decodeBytes(byte[] srcBytes, Context context) {
+    public SourceData decode(byte[] srcBytes, Context context) {
         throw new UnsupportedOperationException("do not support decoding bytes for row data decoder");
     }
 

inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/SourceDecoder.java

@@ -41,7 +41,7 @@ public SourceDecoder(List<FieldInfo> fields) {
         this.fields = fields;
     }
 
-    public abstract SourceData decodeBytes(byte[] srcBytes, Context context);
+    public abstract SourceData decode(byte[] srcBytes, Context context);
 
     public abstract SourceData decode(Input input, Context context);
 

inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/XmlSourceDecoder.java

@@ -58,7 +58,7 @@ public XmlSourceDecoder(XmlSourceInfo sourceInfo) {
     }
 
     @Override
-    public SourceData decodeBytes(byte[] srcBytes, Context context) {
+    public SourceData decode(byte[] srcBytes, Context context) {
         String srcString = new String(srcBytes, srcCharset);
         return this.decode(srcString, context);
     }

inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/YamlSourceDecoder.java

@@ -52,7 +52,7 @@ public YamlSourceDecoder(YamlSourceInfo sourceInfo) {
         }
     }
     @Override
-    public SourceData decodeBytes(byte[] srcBytes, Context context) {
+    public SourceData decode(byte[] srcBytes, Context context) {
         String srcString = new String(srcBytes, srcCharset);
         return this.decode(srcString, context);
     }

inlong-sdk/transform-sdk/src/test/java/org/apache/inlong/sdk/transform/process/processor/AbstractProcessorTestBase.java

@@ -17,6 +17,7 @@
 
 package org.apache.inlong.sdk.transform.process.processor;
 
+import org.apache.inlong.common.pojo.sort.dataflow.field.format.StringFormatInfo;
 import org.apache.inlong.sdk.transform.decode.ParquetInputByteArray;
 import org.apache.inlong.sdk.transform.pojo.FieldInfo;
 
@@ -48,6 +49,7 @@ protected List<FieldInfo> getTestFieldList(String... fieldNames) {
         for (String fieldName : fieldNames) {
             FieldInfo field = new FieldInfo();
             field.setName(fieldName);
+            field.setFormatInfo(new StringFormatInfo());
             fields.add(field);
         }
         return fields;

inlong-sdk/transform-sdk/src/test/java/org/apache/inlong/sdk/transform/process/processor/TestRowData2RowDataProcessor.java

@@ -0,0 +1,71 @@
+/*
+ * 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.sdk.transform.process.processor;
+
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.inlong.sdk.transform.decode.SourceDecoderFactory;
+import org.apache.inlong.sdk.transform.encode.SinkEncoderFactory;
+import org.apache.inlong.sdk.transform.pojo.FieldInfo;
+import org.apache.inlong.sdk.transform.pojo.RowDataSinkInfo;
+import org.apache.inlong.sdk.transform.pojo.RowDataSourceInfo;
+import org.apache.inlong.sdk.transform.pojo.TransformConfig;
+import org.apache.inlong.sdk.transform.process.TransformProcessor;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.List;
+
+public class TestRowData2RowDataProcessor extends AbstractProcessorTestBase {
+
+    @Test
+    public void testRowData2RowData() throws Exception {
+        List<FieldInfo> fields1 = this.getTestFieldList("sid", "packageID", "msgTime", "msg");
+        RowDataSourceInfo sourceInfo = new RowDataSourceInfo("utf-8", fields1);
+        List<FieldInfo> fields2 = this.getTestFieldList("f1", "f2", "f3", "f4");
+        RowDataSinkInfo sinkInfo = new RowDataSinkInfo("utf-8", fields2);
+
+        String transformSql = "select msgTime ,msg, packageID, sid";
+        TransformConfig config = new TransformConfig(transformSql);
+        TransformProcessor<RowData, RowData> processor =
+                TransformProcessor.create(
+                        config,
+                        SourceDecoderFactory.createRowDecoder(sourceInfo),
+                        SinkEncoderFactory.createRowEncoder(sinkInfo));
+
+        RowData sourceRow = createRowData();
+
+        List<RowData> sinkRow = processor.transform(sourceRow);
+        RowData expectedRow = sinkRow.get(0);
+        Assert.assertEquals("2024-12-19T11:00:55.212", expectedRow.getString(0).toString());
+        Assert.assertEquals("msg111", expectedRow.getString(1).toString());
+        Assert.assertEquals("pack123", expectedRow.getString(2).toString());
+        Assert.assertEquals("s123", expectedRow.getString(3).toString());
+
+    }
+
+    private RowData createRowData() {
+        GenericRowData rowData = new GenericRowData(4);
+        rowData.setField(0, StringData.fromString("s123"));
+        rowData.setField(1, StringData.fromString("pack123"));
+        rowData.setField(2, StringData.fromString("2024-12-19T11:00:55.212"));
+        rowData.setField(3, StringData.fromString("msg111"));
+        return rowData;
+    }
+}