NIFI-16074 - PutIcebergRecord missing convertors

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

@@ -19,7 +19,6 @@
 import org.apache.iceberg.data.Record;
 import org.apache.iceberg.types.Types;
 import org.apache.nifi.serialization.record.MapRecord;
-import org.apache.nifi.serialization.record.RecordField;
 
 import java.util.Collections;
 import java.util.LinkedHashMap;
@@ -38,7 +37,7 @@ public DelegatedRecord(
             final org.apache.nifi.serialization.record.Record record,
             final Types.StructType struct
     ) {
-        this.record = RecordConverter.getConvertedRecord(Objects.requireNonNull(record));
+        this.record = RecordConverter.getConvertedRecord(Objects.requireNonNull(record), struct);
         this.struct = Objects.requireNonNull(struct);
     }
 
@@ -77,8 +76,8 @@ public void setField(final String fieldName, final Object fieldValue) {
      */
     @Override
     public Object get(final int position) {
-        final RecordField recordField = record.getSchema().getField(position);
-        return record.getValue(recordField);
+        final String fieldName = struct.fields().get(position).name();
+        return record.getValue(fieldName);
     }
 
     /**
@@ -141,8 +140,8 @@ public <T> T get(final int position, final Class<T> valueClass) {
      */
     @Override
     public <T> void set(final int position, final T value) {
-        final RecordField recordField = record.getSchema().getField(position);
-        record.setValue(recordField, value);
+        final String fieldName = struct.fields().get(position).name();
+        record.setValue(fieldName, value);
     }
 
     @Override

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

@@ -16,20 +16,33 @@
  */
 package org.apache.nifi.processors.iceberg.record;
 
+import org.apache.commons.lang3.ArrayUtils;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.variants.Variant;
+import org.apache.iceberg.variants.VariantPrimitive;
+import org.apache.iceberg.variants.Variants;
 import org.apache.nifi.serialization.record.DataType;
 import org.apache.nifi.serialization.record.MapRecord;
 import org.apache.nifi.serialization.record.Record;
 import org.apache.nifi.serialization.record.RecordField;
 import org.apache.nifi.serialization.record.RecordFieldType;
 import org.apache.nifi.serialization.record.RecordSchema;
 
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
+import java.sql.Timestamp;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
 import java.sql.Date;
 import java.sql.Time;
-import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.UUID;
 
 /**
  * Record Converter handles translating field values to types compatible with Apache Iceberg Records
@@ -39,7 +52,11 @@
     private static final Set<RecordFieldType> CONVERSION_REQUIRED_FIELD_TYPES = Set.of(
             RecordFieldType.TIMESTAMP,
             RecordFieldType.DATE,
-            RecordFieldType.TIME
+            RecordFieldType.TIME,
+            RecordFieldType.UUID,
+            RecordFieldType.ARRAY,
+            RecordFieldType.RECORD,
+            RecordFieldType.MAP
     );
 
     /**
@@ -48,44 +65,116 @@
      * @param inputRecord Input Record to be converted
      * @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);
+        if (isConversionRequired(inputRecord.getSchema(), struct)) {
+            convertedRecord = convertRecord(inputRecord, struct);
         } else {
             convertedRecord = inputRecord;
         }
 
         return convertedRecord;
     }
 
-    private static Record getConvertedRecord(final RecordSchema recordSchema, final Map<String, Object> values) {
+    static Record convertRecord(final Record inputRecord, final Types.StructType struct) {
+        final RecordSchema recordSchema = inputRecord.getSchema();
+        final Map<String, Object> values = inputRecord.toMap();
         final Map<String, Object> convertedValues = new LinkedHashMap<>();
 
-        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);
+        for (final RecordField field : recordSchema.getFields()) {
+            final String fieldName = field.getFieldName();
+            final Types.NestedField icebergField = struct.field(fieldName);
+            final Object rawValue = values.get(fieldName);
+            if (icebergField == null) {
+                convertedValues.put(fieldName, rawValue);
+            }
+            else {
+                convertedValues.put(fieldName, convertValue(rawValue, icebergField.type()));
+            }
         }
 
         return new MapRecord(recordSchema, convertedValues);
     }
 
-    private static Object getConvertedValue(final Object value) {
-        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 null, default -> value;
+    private static Object convertValue(final Object value, final org.apache.iceberg.types.Type icebergType) {
+        if (value == null) {
+            return null;
+        }
+
+        return switch (icebergType.typeId()) {
+            case TIMESTAMP -> {
+                final LocalDateTime local = ((Timestamp) value).toLocalDateTime();
+                yield ((Types.TimestampType) icebergType).shouldAdjustToUTC()
+                        ? local.atOffset(ZoneOffset.UTC)
+                        : local;
+            }
+            case UUID -> value instanceof java.util.UUID ? value : java.util.UUID.fromString(value.toString());
+            case FIXED -> toByteArray(value);
+            case BINARY -> ByteBuffer.wrap(toByteArray(value));
+            case LIST -> {
+                final Iterable<?> elements = value instanceof Object[] array ? Arrays.asList(array) : (List<?>) value;
+                final List<Object> list = new ArrayList<>();
+                for (final Object element : elements) {
+                    list.add(convertValue(element, ((Types.ListType) icebergType).elementType()));
+                }
+                yield list;
+            }
+            case STRUCT -> value instanceof final org.apache.iceberg.data.Record convertedStruct
+                    ? convertedStruct
+                    : new DelegatedRecord(
+                            (org.apache.nifi.serialization.record.Record) value,
+                            (Types.StructType) icebergType
+                    );
+            case MAP -> {
+                final Types.MapType mapType = (Types.MapType) icebergType;
+                final Map<Object, Object> converted = new LinkedHashMap<>();
+                for (final Map.Entry<Object, Object> entry : ((Map<Object, Object>) value).entrySet()) {
+                    final Object convertedKey = convertValue(entry.getKey(), mapType.keyType());
+                    final Object convertedValue = convertValue(entry.getValue(), mapType.valueType());
+                    converted.put(convertedKey, convertedValue);
+                }
+                yield converted;
+            }
+            case DATE -> ((Date) value).toLocalDate();
+            case TIME -> ((Time) value).toLocalTime();
+            case VARIANT -> convertVariant(value);
+            default -> value;
         };
     }
 
-    private static boolean isConversionRequired(final RecordSchema recordSchema) {
+    private static Variant convertVariant(final Object value) {
+        if (value instanceof Variant variant) {
+            return variant;
+        }
+
+        final VariantPrimitive<?> primitive = switch (value) {
+            case Boolean booleanValue -> Variants.of(booleanValue);
+            case Integer integerValue -> Variants.of(integerValue);
+            case Long longValue -> Variants.of(longValue);
+            case Float floatValue -> Variants.of(floatValue);
+            case Double doubleValue -> Variants.of(doubleValue);
+            case BigDecimal decimalValue -> Variants.of(decimalValue);
+            case ByteBuffer byteBufferValue -> Variants.of(byteBufferValue);
+            case byte[] bytesValue -> Variants.of(ByteBuffer.wrap(bytesValue));
+            case UUID uuidValue -> Variants.ofUUID(uuidValue);
+            case String stringValue -> Variants.of(stringValue);
+            default -> Variants.of(value.toString());
+        };
+
+        return Variant.of(Variants.emptyMetadata(), primitive);
+    }
+
+    private static byte[] toByteArray(final Object value) {
+        if (value instanceof byte[] bytes) {
+            return bytes;
+        }
+        final Object[] boxed = (Object[]) value;
+        final Byte[] boxedBytes = Arrays.copyOf(boxed, boxed.length, Byte[].class);
+        return ArrayUtils.toPrimitive(boxedBytes);
+    }
+
+    private static boolean isConversionRequired(final RecordSchema recordSchema, final Types.StructType struct) {
         final List<RecordField> fields = recordSchema.getFields();
 
         for (final RecordField field : fields) {
@@ -94,6 +183,11 @@
             if (CONVERSION_REQUIRED_FIELD_TYPES.contains(recordFieldType)) {
                 return true;
             }
+
+            final Types.NestedField icebergField = struct.field(field.getFieldName());
+            if (icebergField != null && icebergField.type().typeId() == Type.TypeID.VARIANT) {
+                return true;
+            }
         }
 
         return false;

nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/DelegatedRecordTest.java

@@ -82,7 +82,9 @@ void testSetGetStringField() {
 
         final Record record = new MapRecord(recordSchema, values);
 
-        final Types.StructType structType = Types.StructType.of();
+        final Types.StructType structType = Types.StructType.of(
+                Types.NestedField.required(1, LABEL_FIELD, Types.StringType.get())
+        );
         final DelegatedRecord delegatedRecord = new DelegatedRecord(record, structType);
 
         final Types.StructType recordStruct = delegatedRecord.struct();
@@ -123,7 +125,11 @@ void testGetTimestampDateTimeFields() {
 
         final Record record = new MapRecord(recordSchema, values);
 
-        final Types.StructType structType = Types.StructType.of();
+        final Types.StructType structType = Types.StructType.of(
+                Types.NestedField.required(1, CREATED_FIELD, Types.TimestampType.withoutZone()),
+                Types.NestedField.required(2, UPDATED_FIELD, Types.DateType.get()),
+                Types.NestedField.required(3, STOPPED_FIELD, Types.TimeType.get())
+        );
         final DelegatedRecord delegatedRecord = new DelegatedRecord(record, structType);
 
         final Object created = delegatedRecord.getField(CREATED_FIELD);
@@ -135,4 +141,45 @@ void testGetTimestampDateTimeFields() {
         final Object stopped = delegatedRecord.getField(STOPPED_FIELD);
         assertEquals(STOPPED_CONVERTED, stopped);
     }
+
+    @Test
+    void testCopyRecordWithNestedStruct() {
+        final String nestedField = "nested";
+        final String innerTimestampField = "innerTimestamp";
+        final Timestamp innerTimestamp = Timestamp.valueOf("2026-03-04 08:15:30");
+
+        final RecordSchema nestedSchema = new SimpleRecordSchema(
+                List.of(new RecordField(innerTimestampField, RecordFieldType.TIMESTAMP.getDataType()))
+        );
+        final Map<String, Object> nestedValues = new LinkedHashMap<>();
+        nestedValues.put(innerTimestampField, innerTimestamp);
+        final Record nestedRecord = new MapRecord(nestedSchema, nestedValues);
+
+        final RecordSchema outerSchema = new SimpleRecordSchema(
+                List.of(
+                        new RecordField(LABEL_FIELD, RecordFieldType.STRING.getDataType()),
+                        new RecordField(nestedField, RecordFieldType.RECORD.getDataType())
+                )
+        );
+        final Map<String, Object> outerValues = new LinkedHashMap<>();
+        outerValues.put(LABEL_FIELD, LABEL);
+        outerValues.put(nestedField, nestedRecord);
+        final Record outerRecord = new MapRecord(outerSchema, outerValues);
+
+        final Types.StructType nestedStructType = Types.StructType.of(
+                Types.NestedField.required(2, innerTimestampField, Types.TimestampType.withoutZone())
+        );
+        final Types.StructType outerStructType = Types.StructType.of(
+                Types.NestedField.required(1, LABEL_FIELD, Types.StringType.get()),
+                Types.NestedField.required(3, nestedField, nestedStructType)
+        );
+
+        final DelegatedRecord delegatedRecord = new DelegatedRecord(outerRecord, outerStructType);
+
+        final org.apache.iceberg.data.Record firstCopy = delegatedRecord.copy();
+        assertEquals(delegatedRecord, firstCopy);
+
+        final org.apache.iceberg.data.Record secondCopy = ((DelegatedRecord) firstCopy).copy();
+        assertEquals(delegatedRecord, secondCopy);
+    }
 }