[client] enum conversion functionality for typed writer/reader

Author: VladBanar

fluss-client/src/main/java/org/apache/fluss/client/converter/ColumnName.java

@@ -148,6 +148,15 @@ static void validateCompatibility(DataType fieldType, PojoType.Property prop) {
             }
             return;
         }
+        if (actual.isEnum()) {
+            if (typeRoot != DataTypeRoot.STRING) {
+                throw new IllegalArgumentException(
+                        String.format(
+                                "Enum field '%s' must be a string type, got %s",
+                                prop.name, typeRoot));
+            }
+            return;
+        }
 
         Set<Class<?>> supported = SUPPORTED_TYPES.get(fieldType.getTypeRoot());
         if (supported == null) {

fluss-client/src/main/java/org/apache/fluss/client/converter/ConverterCommons.java

@@ -31,9 +31,30 @@
 import java.time.LocalTime;
 import java.time.OffsetDateTime;
 import java.time.ZoneOffset;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 
 /** Shared utilities for Fluss type and Pojo type. */
 public class FlussTypeToPojoTypeConverter {
+
+    private static final Map<Class<?>, Map<String, Object>> ENUM_CONSTANTS_CACHE =
+            new ConcurrentHashMap<>();
+
+    /**
+     * Builds a map of enum name (uppercase) to enum constant for the given enum class. This map is
+     * cached to avoid recreating it for every enum conversion.
+     */
+    private static Map<String, Object> buildEnumConstantsMap(Class<?> enumClass) {
+        Map<String, Object> map = new HashMap<>();
+        for (Object constant : enumClass.getEnumConstants()) {
+            map.put(constant.toString(), constant);
+        }
+        return Collections.unmodifiableMap(map);
+    }
+
     /**
      * Converts a text value (CHAR/STRING) read from an InternalRow into the target Java type
      * declared by the POJO property.
@@ -74,6 +95,18 @@ static Object convertTextValue(
                         ConverterCommons.charLengthExceptionMessage(fieldName, v.length()));
             }
             return v.charAt(0);
+        } else if (pojoType.isEnum()) {
+            Map<String, Object> enumMap =
+                    ENUM_CONSTANTS_CACHE.computeIfAbsent(
+                            pojoType, FlussTypeToPojoTypeConverter::buildEnumConstantsMap);
+            Object enumConstant = enumMap.get(v);
+            if (enumConstant == null) {
+                throw new IllegalArgumentException(
+                        String.format(
+                                "Could not parse value for enum %s. Expected one of: %s",
+                                pojoType, Arrays.toString(pojoType.getEnumConstants())));
+            }
+            return enumConstant;
         }
         throw new IllegalArgumentException(
                 String.format(

fluss-client/src/main/java/org/apache/fluss/client/converter/FlussTypeToPojoTypeConverter.java

@@ -18,6 +18,7 @@
 package org.apache.fluss.client.converter;
 
 import javax.annotation.Nullable;
+
 import java.lang.reflect.Constructor;
 import java.lang.reflect.Field;
 import java.lang.reflect.Method;
@@ -29,8 +30,6 @@
 import java.util.Map;
 import java.util.Objects;
 
-import static org.apache.fluss.utils.Preconditions.checkArgument;
-
 /**
  * Internal representation of a POJO type, used to validate POJO requirements and to provide unified
  * accessors for reading/writing properties.
@@ -74,53 +73,39 @@ static <T> PojoType<T> of(Class<T> pojoClass) {
 
         Map<String, Property> props = new LinkedHashMap<>();
         for (Map.Entry<String, Field> e : allFields.entrySet()) {
-            String fieldName = e.getKey();
+            String name = e.getKey();
             Field field = e.getValue();
             // Enforce nullable fields: primitives are not allowed in POJO definitions.
             if (field.getType().isPrimitive()) {
                 throw new IllegalArgumentException(
                         String.format(
                                 "POJO class %s has primitive field '%s' of type %s. Primitive types are not allowed; all fields must be nullable (use wrapper types).",
-                                pojoClass.getName(), fieldName, field.getType().getName()));
+                                pojoClass.getName(), name, field.getType().getName()));
             }
-            // Check for @ColumnName annotation to determine the mapped column name
-            ColumnName columnNameAnnotation = field.getAnnotation(ColumnName.class);
-            String mappedColumnName =
-                    columnNameAnnotation != null ? columnNameAnnotation.value() : fieldName;
-            checkArgument(
-                    !mappedColumnName.isEmpty(),
-                    "Column name cannot be empty for field '%s' in POJO class %s",
-                    fieldName,
-                    pojoClass.getName());
-
             // use boxed type as effective type
             Class<?> effectiveType = boxIfPrimitive(field.getType());
             boolean publicField = Modifier.isPublic(field.getModifiers());
-            Method getter = getters.get(fieldName);
-            Method setter = setters.get(fieldName);
+            Method getter = getters.get(name);
+            Method setter = setters.get(name);
             if (!publicField) {
                 // When not a public field, require both getter and setter
                 if (getter == null || setter == null) {
-                    final String capitalizedName = capitalize(fieldName);
+                    final String capitalizedName = capitalize(name);
                     throw new IllegalArgumentException(
                             String.format(
                                     "POJO class %s field '%s' must be public or have both getter and setter (get%s/set%s).",
-                                    pojoClass.getName(),
-                                    fieldName,
-                                    capitalizedName,
-                                    capitalizedName));
+                                    pojoClass.getName(), name, capitalizedName, capitalizedName));
                 }
             }
             props.put(
-                    mappedColumnName,
+                    name,
                     new Property(
-                            fieldName,
+                            name,
                             effectiveType,
                             field.getGenericType(),
                             publicField ? field : null,
                             getter,
-                            setter,
-                            mappedColumnName));
+                            setter));
         }
 
         return new PojoType<>(pojoClass, ctor, props);
@@ -284,19 +269,11 @@ private static Class<?> boxIfPrimitive(Class<?> type) {
     }
 
     static final class Property {
-        /** The name of the field in the POJO class (e.g. "userId"). */
         final String name;
-
         final Class<?> type;
         /** The generic type of the field (e.g. {@code Map<String, AddressPojo>}). */
         final Type genericType;
 
-        /**
-         * The name of the column in the Fluss table. This may differ from 'name' if a @ColumnName
-         * annotation is present. Used for looking up the property by table column name.
-         */
-        final String mappedName;
-
         @Nullable final Field publicField;
         @Nullable final Method getter;
         @Nullable final Method setter;
@@ -307,12 +284,10 @@ static final class Property {
                 Type genericType,
                 @Nullable Field publicField,
                 @Nullable Method getter,
-                @Nullable Method setter,
-                String mappedName) {
+                @Nullable Method setter) {
             this.name = Objects.requireNonNull(name, "name");
             this.type = Objects.requireNonNull(type, "type");
             this.genericType = Objects.requireNonNull(genericType, "genericType");
-            this.mappedName = Objects.requireNonNull(mappedName, "mappedName");
             this.publicField = publicField;
             this.getter = getter;
             this.setter = setter;