[fluss-client] Support Complex Data Types on the Java Typed API (NestedRow/ROW)

- Add ROW type support in PojoToRowConverter (POJO -> GenericRow)
- Add ROW type support in RowToPojoConverter (InternalRow -> POJO)
- Add ROW type validation in ConverterCommons
- Add ROW element support in PojoTypeToFlussTypeConverter (for Array/Map)
- Add ROW element support in FlussArrayToPojoArray (for Array<ROW>)
- Add comprehensive tests for nested ROW, deeply nested ROW, ROW with
  Array/Map fields, Array<ROW>, Map<K, ROW>, null handling, and validation
- Update java-client.md documentation with ROW type mapping and usage examples

Author: XuQianJin-Stars

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

@@ -133,6 +133,22 @@ static void validateCompatibility(DataType fieldType, PojoType.Property prop) {
             return;
         }
 
+        if (typeRoot == DataTypeRoot.ROW) {
+            // ROW type maps to a nested POJO. The POJO class must be a valid POJO (public class
+            // with public default constructor). Detailed field-level validation is deferred to
+            // the nested PojoToRowConverter / RowToPojoConverter.
+            if (actual.isPrimitive()
+                    || actual.isArray()
+                    || Collection.class.isAssignableFrom(actual)
+                    || Map.class.isAssignableFrom(actual)) {
+                throw new IllegalArgumentException(
+                        String.format(
+                                "Field '%s' must be a POJO class for ROW type, got %s",
+                                prop.name, actual.getName()));
+            }
+            return;
+        }
+
         Set<Class<?>> supported = SUPPORTED_TYPES.get(fieldType.getTypeRoot());
         if (supported == null) {
             throw new UnsupportedOperationException(

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

@@ -20,11 +20,13 @@
 
 import org.apache.fluss.row.Decimal;
 import org.apache.fluss.row.InternalArray;
+import org.apache.fluss.row.InternalRow;
 import org.apache.fluss.types.ArrayType;
 import org.apache.fluss.types.DataType;
 import org.apache.fluss.types.DataTypeChecks;
 import org.apache.fluss.types.DecimalType;
 import org.apache.fluss.types.MapType;
+import org.apache.fluss.types.RowType;
 
 import java.lang.reflect.Array;
 import java.time.Instant;
@@ -169,6 +171,24 @@ private static ElementConverter buildElementConverter(
                 return (arr, i) ->
                         new FlussMapToPojoMap(arr.getMap(i), (MapType) elementType, fieldName)
                                 .convertMap();
+            case ROW:
+                {
+                    RowType nestedRowType = (RowType) elementType;
+                    int nestedFieldCount = nestedRowType.getFieldCount();
+                    if (pojoType == Object.class) {
+                        // When the target type is unknown (e.g. ROW values in a Map),
+                        // return InternalRow directly
+                        return (arr, i) -> arr.getRow(i, nestedFieldCount);
+                    }
+                    @SuppressWarnings("unchecked")
+                    RowToPojoConverter<Object> nestedConverter =
+                            RowToPojoConverter.of(
+                                    (Class<Object>) pojoType, nestedRowType, nestedRowType);
+                    return (arr, i) -> {
+                        InternalRow nestedRow = arr.getRow(i, nestedFieldCount);
+                        return nestedRow == null ? null : nestedConverter.fromRow(nestedRow);
+                    };
+                }
             default:
                 throw new UnsupportedOperationException(
                         String.format(

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

@@ -157,6 +157,18 @@ private static FieldToRow createFieldConverter(PojoType.Property prop, DataType
                         new PojoMapToFlussMap(
                                         (Map<?, ?>) prop.read(obj), (MapType) fieldType, prop.name)
                                 .convertMap();
+            case ROW:
+                {
+                    RowType nestedRowType = (RowType) fieldType;
+                    @SuppressWarnings("unchecked")
+                    PojoToRowConverter<Object> nestedConverter =
+                            PojoToRowConverter.of(
+                                    (Class<Object>) prop.type, nestedRowType, nestedRowType);
+                    return (obj) -> {
+                        Object nested = prop.read(obj);
+                        return nested == null ? null : nestedConverter.toRow(nested);
+                    };
+                }
             default:
                 throw new UnsupportedOperationException(
                         String.format(

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

@@ -26,6 +26,7 @@
 import org.apache.fluss.types.DataTypeChecks;
 import org.apache.fluss.types.DecimalType;
 import org.apache.fluss.types.MapType;
+import org.apache.fluss.types.RowType;
 
 import javax.annotation.Nullable;
 
@@ -260,6 +261,15 @@ private static int truncateNanos(int nanos, int precision) {
             case MAP:
                 return new PojoMapToFlussMap((Map<?, ?>) obj, (MapType) elementType, fieldName)
                         .convertMap();
+            case ROW:
+                {
+                    RowType nestedRowType = (RowType) elementType;
+                    @SuppressWarnings("unchecked")
+                    PojoToRowConverter<Object> nestedConverter =
+                            PojoToRowConverter.of(
+                                    (Class<Object>) obj.getClass(), nestedRowType, nestedRowType);
+                    return nestedConverter.toRow(obj);
+                }
             default:
                 throw new UnsupportedOperationException(
                         String.format(

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

@@ -196,6 +196,19 @@ private static RowToField createRowReader(DataType fieldType, PojoType.Property
                 return (row, pos) ->
                         new FlussMapToPojoMap(row.getMap(pos), (MapType) fieldType, prop.name)
                                 .convertMap();
+            case ROW:
+                {
+                    RowType nestedRowType = (RowType) fieldType;
+                    int nestedFieldCount = nestedRowType.getFieldCount();
+                    @SuppressWarnings("unchecked")
+                    RowToPojoConverter<Object> nestedConverter =
+                            RowToPojoConverter.of(
+                                    (Class<Object>) prop.type, nestedRowType, nestedRowType);
+                    return (row, pos) -> {
+                        InternalRow nestedRow = row.getRow(pos, nestedFieldCount);
+                        return nestedRow == null ? null : nestedConverter.fromRow(nestedRow);
+                    };
+                }
             default:
                 throw new UnsupportedOperationException(
                         String.format(