From 1aa2c1d6767f4796e68bd761ab3ad345c3d2d015 Mon Sep 17 00:00:00 2001
From: Wenjun Ruan <wenjun@apache.org>
Date: Tue, 2 Jan 2024 11:52:34 +0800
Subject: [PATCH] Fix tiny int convert error in TextSerializationSchema

---
 .../text/TextDeserializationSchema.java       |  7 ++++--
 .../format/text/TextSerializationSchema.java  |  7 +++++-
 .../format/text/TextFormatSchemaTest.java     | 22 +++++++++++++++++++
 3 files changed, 33 insertions(+), 3 deletions(-)

diff --git a/seatunnel-formats/seatunnel-format-text/src/main/java/org/apache/seatunnel/format/text/TextDeserializationSchema.java b/seatunnel-formats/seatunnel-format-text/src/main/java/org/apache/seatunnel/format/text/TextDeserializationSchema.java
index b6429eb600e6..ce981e87e0d7 100644
--- a/seatunnel-formats/seatunnel-format-text/src/main/java/org/apache/seatunnel/format/text/TextDeserializationSchema.java
+++ b/seatunnel-formats/seatunnel-format-text/src/main/java/org/apache/seatunnel/format/text/TextDeserializationSchema.java
@@ -33,6 +33,7 @@
 
 import org.apache.commons.lang3.StringUtils;
 
+import com.google.common.annotations.VisibleForTesting;
 import lombok.NonNull;
 
 import java.io.IOException;
@@ -144,7 +145,8 @@ private Map<Integer, String> splitLineBySeaTunnelRowType(
         return splitsMap;
     }
 
-    private Object convert(String field, SeaTunnelDataType<?> fieldType, int level) {
+    @VisibleForTesting
+    Object convert(String field, SeaTunnelDataType<?> fieldType, int level) {
         if (StringUtils.isBlank(field)) {
             return null;
         }
@@ -201,7 +203,8 @@ private Object convert(String field, SeaTunnelDataType<?> fieldType, int level)
             case BOOLEAN:
                 return Boolean.parseBoolean(field);
             case TINYINT:
-                return Byte.parseByte(field);
+                byte b = Byte.parseByte(field);
+                return b & 0xff;
             case SMALLINT:
                 return Short.parseShort(field);
             case INT:
diff --git a/seatunnel-formats/seatunnel-format-text/src/main/java/org/apache/seatunnel/format/text/TextSerializationSchema.java b/seatunnel-formats/seatunnel-format-text/src/main/java/org/apache/seatunnel/format/text/TextSerializationSchema.java
index e33095703bdc..7d81162fb9b8 100644
--- a/seatunnel-formats/seatunnel-format-text/src/main/java/org/apache/seatunnel/format/text/TextSerializationSchema.java
+++ b/seatunnel-formats/seatunnel-format-text/src/main/java/org/apache/seatunnel/format/text/TextSerializationSchema.java
@@ -31,6 +31,7 @@
 import org.apache.seatunnel.format.text.constant.TextFormatConstant;
 import org.apache.seatunnel.format.text.exception.SeaTunnelTextFormatException;
 
+import com.google.common.annotations.VisibleForTesting;
 import lombok.NonNull;
 
 import java.time.LocalDate;
@@ -124,7 +125,8 @@ public byte[] serialize(SeaTunnelRow element) {
         return String.join(separators[0], strings).getBytes();
     }
 
-    private String convert(Object field, SeaTunnelDataType<?> fieldType, int level) {
+    @VisibleForTesting
+    String convert(Object field, SeaTunnelDataType<?> fieldType, int level) {
         if (field == null) {
             return "";
         }
@@ -135,6 +137,9 @@ private String convert(Object field, SeaTunnelDataType<?> fieldType, int level)
             case STRING:
             case BOOLEAN:
             case TINYINT:
+                int value = (int) field;
+                byte byteValue = (byte) value;
+                return String.valueOf(byteValue);
             case SMALLINT:
             case BIGINT:
             case DECIMAL:
diff --git a/seatunnel-formats/seatunnel-format-text/src/test/java/org/apache/seatunnel/format/text/TextFormatSchemaTest.java b/seatunnel-formats/seatunnel-format-text/src/test/java/org/apache/seatunnel/format/text/TextFormatSchemaTest.java
index 57e99d49b69e..2876b6b5f1bc 100644
--- a/seatunnel-formats/seatunnel-format-text/src/test/java/org/apache/seatunnel/format/text/TextFormatSchemaTest.java
+++ b/seatunnel-formats/seatunnel-format-text/src/test/java/org/apache/seatunnel/format/text/TextFormatSchemaTest.java
@@ -145,4 +145,26 @@ public void testParse() throws IOException {
         Assertions.assertEquals(seaTunnelRow.getField(2), "tyrantlucifer");
         Assertions.assertEquals(data, content);
     }
+
+    @Test
+    public void testConvertTinyInt() {
+        TextDeserializationSchema textDeserializationSchema =
+                TextDeserializationSchema.builder().seaTunnelRowType(seaTunnelRowType).build();
+        TextSerializationSchema textSerializationSchema =
+                TextSerializationSchema.builder().seaTunnelRowType(seaTunnelRowType).build();
+        String tinyIntSerialization = textSerializationSchema.convert(0, BasicType.BYTE_TYPE, 0);
+        Object tinyIntDeserialization =
+                textDeserializationSchema.convert(tinyIntSerialization, BasicType.BYTE_TYPE, 0);
+        Assertions.assertEquals(0, tinyIntDeserialization);
+
+        tinyIntSerialization = textSerializationSchema.convert(122, BasicType.BYTE_TYPE, 0);
+        tinyIntDeserialization =
+                textDeserializationSchema.convert(tinyIntSerialization, BasicType.BYTE_TYPE, 0);
+        Assertions.assertEquals(122, tinyIntDeserialization);
+
+        tinyIntSerialization = textSerializationSchema.convert(255, BasicType.BYTE_TYPE, 0);
+        tinyIntDeserialization =
+                textDeserializationSchema.convert(tinyIntSerialization, BasicType.BYTE_TYPE, 0);
+        Assertions.assertEquals(255, tinyIntDeserialization);
+    }
 }