check the type class for char and character

Author: polyzos

fluss-client/src/main/java/org/apache/fluss/client/utils/PojoConverterUtils.java

@@ -24,8 +24,8 @@
 import org.apache.fluss.row.TimestampLtz;
 import org.apache.fluss.row.TimestampNtz;
 import org.apache.fluss.types.DataType;
-import org.apache.fluss.types.DataTypeRoot;
 import org.apache.fluss.types.DataTypeChecks;
+import org.apache.fluss.types.DataTypeRoot;
 import org.apache.fluss.types.DecimalType;
 import org.apache.fluss.types.RowType;
 
@@ -391,7 +391,26 @@ private RowToFieldConverter createRowToFieldConverter(DataType fieldType, Field
                         return null;
                     }
                     BinaryString binaryString = row.getString(pos);
-                    return binaryString.toString();
+                    String value = binaryString.toString();
+                    if (fieldClass == String.class) {
+                        return value;
+                    } else if (fieldClass == Character.class || fieldClass == char.class) {
+                        if (value.isEmpty()) {
+                            throw new IllegalArgumentException(
+                                    String.format(
+                                            "Field %s expects Character/char, but the string value is empty.",
+                                            field.getName()));
+                        }
+                        return value.charAt(0);
+                    } else {
+                        // This should normally be prevented by constructor-time validation of
+                        // supported types.
+                        // Keep a defensive check here for clarity.
+                        throw new IllegalArgumentException(
+                                String.format(
+                                        "Field %s is not a String or Character/char. Cannot convert from string.",
+                                        field.getName()));
+                    }
                 };
             case BINARY:
             case BYTES:
@@ -422,35 +441,38 @@ private RowToFieldConverter createRowToFieldConverter(DataType fieldType, Field
                     int millis = row.getInt(pos);
                     return LocalTime.ofNanoOfDay(millis * 1_000_000L);
                 };
-            case TIMESTAMP_WITHOUT_TIME_ZONE: {
-                final int precision = DataTypeChecks.getPrecision(fieldType);
-                return (row, pos) -> {
-                    if (row.isNullAt(pos)) {
-                        return null;
-                    }
-                    TimestampNtz timestampNtz = row.getTimestampNtz(pos, precision);
-                    return timestampNtz.toLocalDateTime();
-                };
-            }
-            case TIMESTAMP_WITH_LOCAL_TIME_ZONE: {
-                final int precision = DataTypeChecks.getPrecision(fieldType);
-                return (row, pos) -> {
-                    if (row.isNullAt(pos)) {
-                        return null;
-                    }
-                    TimestampLtz timestampLtz = row.getTimestampLtz(pos, precision);
-                    if (fieldClass == Instant.class) {
-                        return timestampLtz.toInstant();
-                    } else if (fieldClass == OffsetDateTime.class) {
-                        return OffsetDateTime.ofInstant(timestampLtz.toInstant(), ZoneOffset.UTC);
-                    } else {
-                        throw new IllegalArgumentException(
-                                String.format(
-                                        "Field %s is not an Instant or OffsetDateTime. Cannot convert from TimestampData.",
-                                        field.getName()));
-                    }
-                };
-            }
+            case TIMESTAMP_WITHOUT_TIME_ZONE:
+                {
+                    final int precision = DataTypeChecks.getPrecision(fieldType);
+                    return (row, pos) -> {
+                        if (row.isNullAt(pos)) {
+                            return null;
+                        }
+                        TimestampNtz timestampNtz = row.getTimestampNtz(pos, precision);
+                        return timestampNtz.toLocalDateTime();
+                    };
+                }
+            case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+                {
+                    final int precision = DataTypeChecks.getPrecision(fieldType);
+                    return (row, pos) -> {
+                        if (row.isNullAt(pos)) {
+                            return null;
+                        }
+                        TimestampLtz timestampLtz = row.getTimestampLtz(pos, precision);
+                        if (fieldClass == Instant.class) {
+                            return timestampLtz.toInstant();
+                        } else if (fieldClass == OffsetDateTime.class) {
+                            return OffsetDateTime.ofInstant(
+                                    timestampLtz.toInstant(), ZoneOffset.UTC);
+                        } else {
+                            throw new IllegalArgumentException(
+                                    String.format(
+                                            "Field %s is not an Instant or OffsetDateTime. Cannot convert from TimestampData.",
+                                            field.getName()));
+                        }
+                    };
+                }
             default:
                 throw new UnsupportedOperationException(
                         String.format(
@@ -538,7 +560,6 @@ public T fromRow(InternalRow row) {
         }
     }
 
-
     /**
      * Utility method to create a Set containing the specified Java type classes.
      *

fluss-client/src/test/java/org/apache/fluss/client/utils/PojoConverterUtilsTest.java

@@ -360,6 +360,52 @@ public DecimalWrongTypePojo(String decimalField) {
         }
     }
 
+    @Test
+    public void testCharacterAndCharFieldConversionRoundTrip() {
+        // Use STRING schema to test both Character and primitive char fields
+        RowType rowType = RowType.builder().field("charField", DataTypes.STRING()).build();
+
+        // Character wrapper field
+        PojoConverterUtils<CharacterFieldPojo> characterConverter =
+                PojoConverterUtils.getConverter(CharacterFieldPojo.class, rowType);
+        CharacterFieldPojo characterPojo = new CharacterFieldPojo('A');
+        GenericRow row1 = characterConverter.toRow(characterPojo);
+        assertThat(row1.getString(0).toString()).isEqualTo("A");
+        CharacterFieldPojo characterPojo2 = characterConverter.fromRow(row1);
+        assertThat(characterPojo2.charField).isEqualTo('A');
+
+        // primitive char field
+        PojoConverterUtils<PrimitiveCharFieldPojo> primitiveConverter =
+                PojoConverterUtils.getConverter(PrimitiveCharFieldPojo.class, rowType);
+        PrimitiveCharFieldPojo primitivePojo = new PrimitiveCharFieldPojo('Z');
+        GenericRow row2 = primitiveConverter.toRow(primitivePojo);
+        assertThat(row2.getString(0).toString()).isEqualTo("Z");
+        PrimitiveCharFieldPojo primitivePojo2 = primitiveConverter.fromRow(row2);
+        assertThat(primitivePojo2.charField).isEqualTo('Z');
+    }
+
+    /** POJO with a Character field for testing CHAR/STRING conversions to Character. */
+    public static class CharacterFieldPojo {
+        private Character charField;
+
+        public CharacterFieldPojo() {}
+
+        public CharacterFieldPojo(Character charField) {
+            this.charField = charField;
+        }
+    }
+
+    /** POJO with a primitive char field for testing CHAR/STRING conversions to char. */
+    public static class PrimitiveCharFieldPojo {
+        private char charField;
+
+        public PrimitiveCharFieldPojo() {}
+
+        public PrimitiveCharFieldPojo(char charField) {
+            this.charField = charField;
+        }
+    }
+
     /** Helper POJO to trigger DATE field type mismatch. */
     public static class DateWrongTypePojo {
         private String dateField;