NIFI-16069 - PutIcebergRecord fails with ClassCastException when writing complex types (arrays, maps, nested records)

nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-parquet-writer/src/test/java/org/apache/nifi/services/iceberg/parquet/ParquetIcebergWriterTest.java

@@ -47,6 +47,8 @@
 import java.time.LocalDate;
 import java.time.LocalDateTime;
 import java.time.LocalTime;
+import java.util.List;
+import java.util.Map;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertInstanceOf;
@@ -76,6 +78,24 @@ class ParquetIcebergWriterTest {
             LocalDate.ofEpochDay(1), LocalTime.ofSecondOfDay(0)
     );
 
+    private static final String ID_FIELD_NAME = "id";
+
+    private static final String TAGS_FIELD_NAME = "tags";
+
+    private static final String ADDRESS_FIELD_NAME = "address";
+
+    private static final String CITY_FIELD_NAME = "city";
+
+    private static final String ATTRIBUTES_FIELD_NAME = "attributes";
+
+    private static final String ID_FIELD_VALUE = "row-1";
+
+    private static final String CITY_FIELD_VALUE = "Berlin";
+
+    private static final List<String> TAGS_FIELD_VALUE = List.of("a", "b");
+
+    private static final Map<String, String> ATTRIBUTES_FIELD_VALUE = Map.of("k", "v");
+
     private ParquetIcebergWriter parquetIcebergWriter;
 
     private TestRunner runner;
@@ -194,6 +214,43 @@ void testWriteDataFilesPartitionedTimestamp() throws IOException {
         assertEquals(microsecondsExpected, partitionField);
     }
 
+    @Test
+    void testWriteDataFilesComplexTypes() throws IOException {
+        runner.enableControllerService(parquetIcebergWriter);
+
+        final Types.StructType nestedStruct = Types.StructType.of(
+                Types.NestedField.optional(10, CITY_FIELD_NAME, Types.StringType.get())
+        );
+        final Schema schema = new Schema(
+                Types.NestedField.required(1, ID_FIELD_NAME, Types.StringType.get()),
+                Types.NestedField.optional(2, TAGS_FIELD_NAME,
+                        Types.ListType.ofOptional(3, Types.StringType.get())),
+                Types.NestedField.optional(4, ADDRESS_FIELD_NAME, nestedStruct),
+                Types.NestedField.optional(5, ATTRIBUTES_FIELD_NAME,
+                        Types.MapType.ofOptional(6, 7, Types.StringType.get(), Types.StringType.get()))
+        );
+        final InMemoryOutputFile outputFile = new InMemoryOutputFile();
+        final PartitionSpec partitionSpec = PartitionSpec.unpartitioned();
+        setTable(schema, partitionSpec, outputFile);
+        when(locationProvider.newDataLocation(anyString())).thenReturn(LOCATION);
+
+        final IcebergRowWriter rowWriter = parquetIcebergWriter.getRowWriter(table);
+
+        final GenericRecord address = GenericRecord.create(nestedStruct);
+        address.setField(CITY_FIELD_NAME, CITY_FIELD_VALUE);
+
+        final GenericRecord row = GenericRecord.create(schema);
+        row.setField(ID_FIELD_NAME, ID_FIELD_VALUE);
+        row.setField(TAGS_FIELD_NAME, TAGS_FIELD_VALUE);
+        row.setField(ADDRESS_FIELD_NAME, address);
+        row.setField(ATTRIBUTES_FIELD_NAME, ATTRIBUTES_FIELD_VALUE);
+        rowWriter.write(row);
+
+        final DataFile[] dataFiles = rowWriter.dataFiles();
+        final byte[] serialized = outputFile.toByteArray();
+        assertDataFilesFound(dataFiles, serialized);
+    }
+
     private void writeRow(final Schema schema, final IcebergRowWriter rowWriter) throws IOException {
         final GenericRecord row = GenericRecord.create(schema);
         row.setField(FIRST_FIELD_NAME, FIRST_FIELD_VALUE);

nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/DelegatedRecord.java

@@ -37,8 +37,8 @@ public DelegatedRecord(
             final org.apache.nifi.serialization.record.Record record,
             final Types.StructType struct
     ) {
-        this.record = RecordConverter.getConvertedRecord(Objects.requireNonNull(record));
         this.struct = Objects.requireNonNull(struct);
+        this.record = RecordConverter.getConvertedRecord(Objects.requireNonNull(record), struct);
     }
 
     @Override

nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java

@@ -16,6 +16,8 @@
  */
 package org.apache.nifi.processors.iceberg.record;
 
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.Types;
 import org.apache.nifi.serialization.record.DataType;
 import org.apache.nifi.serialization.record.MapRecord;
 import org.apache.nifi.serialization.record.Record;
@@ -26,6 +28,10 @@
 import java.sql.Date;
 import java.sql.Time;
 import java.sql.Timestamp;
+import java.time.ZoneOffset;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
@@ -39,63 +45,151 @@ class RecordConverter {
     private static final Set<RecordFieldType> CONVERSION_REQUIRED_FIELD_TYPES = Set.of(
             RecordFieldType.TIMESTAMP,
             RecordFieldType.DATE,
-            RecordFieldType.TIME
+            RecordFieldType.TIME,
+            RecordFieldType.ARRAY,
+            RecordFieldType.RECORD,
+            RecordFieldType.MAP,
+            // CHOICE can wrap any of the above, so it must also trigger conversion.
+            RecordFieldType.CHOICE
     );
 
     /**
-     * Get Converted Record with conditional handling for field values requiring translation
+     * Get Converted Record with recursive, schema-aware handling for field values requiring translation
      *
      * @param inputRecord Input Record to be converted
+     * @param struct Iceberg Struct Type describing the target field types (may be null for scalar-only conversion)
      * @return Input Record or new Record with converted field values
      */
-    static Record getConvertedRecord(final Record inputRecord) {
+    static Record getConvertedRecord(final Record inputRecord, final Types.StructType struct) {
         final Record convertedRecord;
 
         final RecordSchema recordSchema = inputRecord.getSchema();
         if (isConversionRequired(recordSchema)) {
             final Map<String, Object> values = inputRecord.toMap();
-            convertedRecord = getConvertedRecord(recordSchema, values);
+            final Map<String, Object> convertedValues = new LinkedHashMap<>(values.size());
+            for (final Map.Entry<String, Object> entry : values.entrySet()) {
+                final String field = entry.getKey();
+                final Type fieldType = fieldType(struct, field);
+                convertedValues.put(field, convertValue(entry.getValue(), fieldType));
+            }
+            convertedRecord = new MapRecord(recordSchema, convertedValues);
         } else {
             convertedRecord = inputRecord;
         }
 
         return convertedRecord;
     }
 
-    private static Record getConvertedRecord(final RecordSchema recordSchema, final Map<String, Object> values) {
-        final Map<String, Object> convertedValues = new LinkedHashMap<>();
+    static Object convertValue(final Object value, final Type icebergType) {
+        return switch (value) {
+            // Convert java.sql types to corresponding java.time types for Apache Iceberg
+            case Timestamp timestamp -> convertTimestamp(timestamp, icebergType);
+            case Date date -> date.toLocalDate();
+            case Time time -> time.toLocalTime();
+            // Recursively convert complex types against the matching Iceberg type
+            case null, default -> convertComplexValue(value, icebergType);
+        };
+    }
 
-        for (final Map.Entry<String, Object> entry : values.entrySet()) {
-            final String field = entry.getKey();
-            final Object value = entry.getValue();
-            final Object converted = getConvertedValue(value);
-            convertedValues.put(field, converted);
+    /**
+     * Convert a Timestamp to the java.time type required by the target Iceberg Type. Iceberg Types declaring an
+     * adjustment to UTC require an OffsetDateTime, and other Types require a LocalDateTime. A Timestamp identifies
+     * an instant, so the adjusted conversion preserves that instant expressed at UTC
+     *
+     * @param timestamp Timestamp to be converted
+     * @param icebergType Iceberg Type describing the target field type (may be null when not resolved)
+     * @return OffsetDateTime at UTC for Iceberg Types adjusted to UTC or LocalDateTime for other Types
+     */
+    private static Object convertTimestamp(final Timestamp timestamp, final Type icebergType) {
+        return shouldAdjustToUtc(icebergType) ? timestamp.toInstant().atOffset(ZoneOffset.UTC) : timestamp.toLocalDateTime();
+    }
+
+    /**
+     * Determine whether the Iceberg Type declares an adjustment to UTC, which Apache Iceberg requires for the
+     * timestamptz and timestamptz_ns column types
+     *
+     * @param icebergType Iceberg Type describing the target field type (may be null when not resolved)
+     * @return Adjustment to UTC required status
+     */
+    private static boolean shouldAdjustToUtc(final Type icebergType) {
+        return switch (icebergType) {
+            case Types.TimestampType timestampType -> timestampType.shouldAdjustToUTC();
+            case Types.TimestampNanoType timestampNanoType -> timestampNanoType.shouldAdjustToUTC();
+            case null, default -> false;
+        };
+    }
+
+    /**
+     * Recursively convert array, collection, nested record, and map values against the matching Iceberg type
+     *
+     * @param value Field value to be converted
+     * @param icebergType Iceberg Type describing the target field type (may be null when not resolved)
+     * @return Converted value or the input value when the Iceberg Type is unknown or does not describe a complex
+     * type matching the value
+     */
+    private static Object convertComplexValue(final Object value, final Type icebergType) {
+        final Object convertedValue;
+
+        if (icebergType == null) {
+            convertedValue = value;
+        } else if (icebergType.isListType()) {
+            convertedValue = convertListValue(value, icebergType.asListType());
+        } else if (icebergType.isStructType() && value instanceof Record nestedRecord) {
+            convertedValue = new DelegatedRecord(nestedRecord, icebergType.asStructType());
+        } else if (icebergType.isMapType() && value instanceof Map<?, ?> map) {
+            convertedValue = convertMap(map, icebergType.asMapType());
+        } else {
+            convertedValue = value;
         }
 
-        return new MapRecord(recordSchema, convertedValues);
+        return convertedValue;
     }
 
-    private static Object getConvertedValue(final Object value) {
+    /**
+     * Convert an array or collection value to the List required for Apache Iceberg with elements converted against
+     * the Iceberg element type
+     *
+     * @param value Field value to be converted
+     * @param listType Iceberg List Type describing the target element type
+     * @return Converted List or the input value when the value is neither an array nor a collection
+     */
+    private static Object convertListValue(final Object value, final Types.ListType listType) {
+        final Type elementType = listType.elementType();
         return switch (value) {
-            // Convert java.sql types to corresponding java.time types for Apache Iceberg
-            case Timestamp timestamp -> timestamp.toLocalDateTime();
-            case Date date -> date.toLocalDate();
-            case Time time -> time.toLocalTime();
+            case Object[] array -> convertList(Arrays.asList(array), elementType);
+            case Collection<?> collection -> convertList(collection, elementType);
             case null, default -> value;
         };
     }
 
-    private static boolean isConversionRequired(final RecordSchema recordSchema) {
-        final List<RecordField> fields = recordSchema.getFields();
+    private static List<Object> convertList(final Collection<?> collection, final Type elementType) {
+        final List<Object> converted = new ArrayList<>(collection.size());
+        for (final Object element : collection) {
+            converted.add(convertValue(element, elementType));
+        }
+        return converted;
+    }
 
-        for (final RecordField field : fields) {
-            final DataType dataType = field.getDataType();
-            final RecordFieldType recordFieldType = dataType.getFieldType();
-            if (CONVERSION_REQUIRED_FIELD_TYPES.contains(recordFieldType)) {
-                return true;
-            }
+    private static Map<Object, Object> convertMap(final Map<?, ?> map, final Types.MapType mapType) {
+        // Using LinkedHashMap here to keep input ordering for deterministic flows.
+        final Map<Object, Object> converted = new LinkedHashMap<>(map.size());
+        for (final Map.Entry<?, ?> entry : map.entrySet()) {
+            final Object key = convertValue(entry.getKey(), mapType.keyType());
+            final Object mappedValue = convertValue(entry.getValue(), mapType.valueType());
+            converted.put(key, mappedValue);
         }
+        return converted;
+    }
+
+    private static Type fieldType(final Types.StructType struct, final String fieldName) {
+        final Types.NestedField nestedField = struct == null ? null : struct.field(fieldName);
+        return nestedField == null ? null : nestedField.type();
+    }
 
-        return false;
+    private static boolean isConversionRequired(final RecordSchema recordSchema) {
+        return recordSchema.getFields().stream()
+                .map(RecordField::getDataType)
+                .map(DataType::getFieldType)
+                .anyMatch(CONVERSION_REQUIRED_FIELD_TYPES::contains);
     }
 }