From 14ba86171614a713f51e02b0edfa63d2204206f5 Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Thu, 2 Jul 2026 19:24:21 +0200 Subject: [PATCH 1/9] NIFI-16069 - PutIcebergRecord fails with ClassCastException when writing complex types (arrays, maps, nested records) --- .../parquet/ParquetIcebergWriterTest.java | 39 +++++ .../iceberg/record/DelegatedRecord.java | 2 +- .../iceberg/record/RecordConverter.java | 80 ++++++--- .../iceberg/record/RecordConverterTest.java | 162 ++++++++++++++++++ 4 files changed, 256 insertions(+), 27 deletions(-) create mode 100644 nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/RecordConverterTest.java diff --git a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-parquet-writer/src/test/java/org/apache/nifi/services/iceberg/parquet/ParquetIcebergWriterTest.java b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-parquet-writer/src/test/java/org/apache/nifi/services/iceberg/parquet/ParquetIcebergWriterTest.java index 636b0ba091f1..b0159191fe44 100644 --- a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-parquet-writer/src/test/java/org/apache/nifi/services/iceberg/parquet/ParquetIcebergWriterTest.java +++ b/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; @@ -194,6 +196,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", Types.StringType.get()) + ); + final Schema schema = new Schema( + Types.NestedField.required(1, "id", Types.StringType.get()), + Types.NestedField.optional(2, "tags", + Types.ListType.ofOptional(3, Types.StringType.get())), + Types.NestedField.optional(4, "address", nestedStruct), + Types.NestedField.optional(5, "attributes", + 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", "Berlin"); + + final GenericRecord row = GenericRecord.create(schema); + row.setField("id", "row-1"); + row.setField("tags", List.of("a", "b")); + row.setField("address", address); + row.setField("attributes", Map.of("k", "v")); + 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); diff --git a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/DelegatedRecord.java b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/DelegatedRecord.java index eff7652be1ae..565277cebf77 100644 --- a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/DelegatedRecord.java +++ b/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 diff --git a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java index b84159d4af0a..4c8f6e8f567c 100644 --- a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java +++ b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java @@ -16,7 +16,8 @@ */ package org.apache.nifi.processors.iceberg.record; -import org.apache.nifi.serialization.record.DataType; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.Types; import org.apache.nifi.serialization.record.MapRecord; import org.apache.nifi.serialization.record.Record; import org.apache.nifi.serialization.record.RecordField; @@ -26,6 +27,9 @@ import java.sql.Date; import java.sql.Time; import java.sql.Timestamp; +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 +43,87 @@ class RecordConverter { private static final Set CONVERSION_REQUIRED_FIELD_TYPES = Set.of( RecordFieldType.TIMESTAMP, RecordFieldType.DATE, - RecordFieldType.TIME + RecordFieldType.TIME, + RecordFieldType.ARRAY, + RecordFieldType.RECORD, + RecordFieldType.MAP ); /** - * 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) { - final Record convertedRecord; - + static Record getConvertedRecord(final Record inputRecord, final Types.StructType struct) { final RecordSchema recordSchema = inputRecord.getSchema(); - if (isConversionRequired(recordSchema)) { - final Map values = inputRecord.toMap(); - convertedRecord = getConvertedRecord(recordSchema, values); - } else { - convertedRecord = inputRecord; + if (!isConversionRequired(recordSchema)) { + return inputRecord; } - return convertedRecord; - } - - private static Record getConvertedRecord(final RecordSchema recordSchema, final Map values) { + final Map values = inputRecord.toMap(); final Map convertedValues = new LinkedHashMap<>(); - for (final Map.Entry entry : values.entrySet()) { final String field = entry.getKey(); - final Object value = entry.getValue(); - final Object converted = getConvertedValue(value); - convertedValues.put(field, converted); + final Type fieldType = fieldType(struct, field); + convertedValues.put(field, convertValue(entry.getValue(), fieldType)); } return new MapRecord(recordSchema, convertedValues); } - private static Object getConvertedValue(final Object value) { + 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 -> timestamp.toLocalDateTime(); case Date date -> date.toLocalDate(); case Time time -> time.toLocalTime(); + case Object[] array when icebergType != null && icebergType.isListType() -> + convertList(Arrays.asList(array), icebergType.asListType().elementType()); + case Collection collection when icebergType != null && icebergType.isListType() -> + convertList(collection, icebergType.asListType().elementType()); + case Record nestedRecord when icebergType != null && icebergType.isStructType() -> + new DelegatedRecord(nestedRecord, icebergType.asStructType()); + case Map map when icebergType != null && icebergType.isMapType() -> + convertMap(map, icebergType.asMapType()); case null, default -> value; }; } - private static boolean isConversionRequired(final RecordSchema recordSchema) { - final List fields = recordSchema.getFields(); + private static List convertList(final Collection collection, final Type elementType) { + final List 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(); + private static Map convertMap(final Map map, final Types.MapType mapType) { + final Map converted = new LinkedHashMap<>(); + 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) { + if (struct == null) { + return null; + } + final Types.NestedField nestedField = struct.field(fieldName); + return nestedField == null ? null : nestedField.type(); + } + + private static boolean isConversionRequired(final RecordSchema recordSchema) { + for (final RecordField field : recordSchema.getFields()) { + final RecordFieldType recordFieldType = field.getDataType().getFieldType(); if (CONVERSION_REQUIRED_FIELD_TYPES.contains(recordFieldType)) { return true; } } - return false; } } diff --git a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/RecordConverterTest.java b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/RecordConverterTest.java new file mode 100644 index 000000000000..0b77598b2a09 --- /dev/null +++ b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/RecordConverterTest.java @@ -0,0 +1,162 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.processors.iceberg.record; + +import org.apache.iceberg.StructLike; +import org.apache.iceberg.types.Types; +import org.apache.nifi.serialization.SimpleRecordSchema; +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 org.junit.jupiter.api.Test; + +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; + +class RecordConverterTest { + + @Test + void testConvertPrimitiveArrayToList() { + final Types.ListType listType = Types.ListType.ofOptional(1, Types.StringType.get()); + final Object[] array = new Object[] {"a", "b", "c"}; + + final Object converted = RecordConverter.convertValue(array, listType); + + final List list = assertInstanceOf(List.class, converted); + assertEquals(List.of("a", "b", "c"), list); + } + + @Test + void testConvertArrayElementDateTime() { + final Types.ListType listType = Types.ListType.ofOptional(1, Types.DateType.get()); + final Object[] array = new Object[] {java.sql.Date.valueOf("2026-02-03")}; + + final Object converted = RecordConverter.convertValue(array, listType); + + final List list = assertInstanceOf(List.class, converted); + assertEquals(List.of(LocalDate.of(2026, 2, 3)), list); + } + + @Test + void testConvertNestedRecordToStructLike() { + final Types.StructType structType = Types.StructType.of( + Types.NestedField.optional(1, "city", Types.StringType.get()) + ); + + final RecordSchema nestedSchema = new SimpleRecordSchema(List.of( + new RecordField("city", RecordFieldType.STRING.getDataType()) + )); + final Map nestedValues = new LinkedHashMap<>(); + nestedValues.put("city", "Berlin"); + final Record nestedRecord = new MapRecord(nestedSchema, nestedValues); + + final Object converted = RecordConverter.convertValue(nestedRecord, structType); + + final StructLike struct = assertInstanceOf(StructLike.class, converted); + assertEquals("Berlin", struct.get(0, String.class)); + } + + @Test + void testConvertNestedRecordDateTimeField() { + final Types.StructType structType = Types.StructType.of( + Types.NestedField.optional(1, "created", Types.TimestampType.withoutZone()) + ); + + final RecordSchema nestedSchema = new SimpleRecordSchema(List.of( + new RecordField("created", RecordFieldType.TIMESTAMP.getDataType()) + )); + final Map nestedValues = new LinkedHashMap<>(); + nestedValues.put("created", java.sql.Timestamp.valueOf("2026-01-01 12:30:45")); + final Record nestedRecord = new MapRecord(nestedSchema, nestedValues); + + final Object converted = RecordConverter.convertValue(nestedRecord, structType); + + final StructLike struct = assertInstanceOf(StructLike.class, converted); + assertEquals(LocalDateTime.of(2026, 1, 1, 12, 30, 45), struct.get(0, LocalDateTime.class)); + } + + @Test + void testConvertMapValues() { + final Types.MapType mapType = Types.MapType.ofOptional( + 1, 2, Types.StringType.get(), Types.StringType.get() + ); + final Map map = new LinkedHashMap<>(); + map.put("k1", "v1"); + map.put("k2", "v2"); + + final Object converted = RecordConverter.convertValue(map, mapType); + + final Map resultMap = assertInstanceOf(Map.class, converted); + assertEquals("v1", resultMap.get("k1")); + assertEquals("v2", resultMap.get("k2")); + } + + @Test + void testConvertMapDateTimeValue() { + final Types.MapType mapType = Types.MapType.ofOptional( + 1, 2, Types.StringType.get(), Types.DateType.get() + ); + final Map map = new LinkedHashMap<>(); + map.put("day", java.sql.Date.valueOf("2026-02-03")); + + final Object converted = RecordConverter.convertValue(map, mapType); + + final Map resultMap = assertInstanceOf(Map.class, converted); + assertEquals(LocalDate.of(2026, 2, 3), resultMap.get("day")); + } + + @Test + void testGetConvertedRecordArrayOfStructs() { + final Types.StructType elementStruct = Types.StructType.of( + Types.NestedField.optional(2, "name", Types.StringType.get()) + ); + final Types.StructType struct = Types.StructType.of( + Types.NestedField.optional(1, "items", + Types.ListType.ofOptional(3, elementStruct)) + ); + + final RecordSchema elementSchema = new SimpleRecordSchema(List.of( + new RecordField("name", RecordFieldType.STRING.getDataType()) + )); + final Map elementValues = new LinkedHashMap<>(); + elementValues.put("name", "widget"); + final Record element = new MapRecord(elementSchema, elementValues); + + final RecordSchema schema = new SimpleRecordSchema(List.of( + new RecordField("items", + RecordFieldType.ARRAY.getArrayDataType(RecordFieldType.RECORD.getRecordDataType(elementSchema))) + )); + final Map values = new LinkedHashMap<>(); + values.put("items", new Object[] {element}); + final Record record = new MapRecord(schema, values); + + final org.apache.iceberg.data.Record converted = new DelegatedRecord(record, struct); + final Object items = converted.getField("items"); + + final List list = assertInstanceOf(List.class, items); + final StructLike first = assertInstanceOf(StructLike.class, list.get(0)); + assertEquals("widget", first.get(0, String.class)); + } +} From 5659f3abbe0d6de3cb965a9d6c2fac311a3d981d Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Fri, 10 Jul 2026 22:24:33 +0200 Subject: [PATCH 2/9] refactor: extract nested complex conditionals to convertComplexValue method --- .../iceberg/record/RecordConverter.java | 38 ++++++++++++++----- 1 file changed, 29 insertions(+), 9 deletions(-) diff --git a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java index 4c8f6e8f567c..4cddb93b4ac0 100644 --- a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java +++ b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java @@ -79,18 +79,38 @@ static Object convertValue(final Object value, final Type icebergType) { case Timestamp timestamp -> timestamp.toLocalDateTime(); case Date date -> date.toLocalDate(); case Time time -> time.toLocalTime(); - case Object[] array when icebergType != null && icebergType.isListType() -> - convertList(Arrays.asList(array), icebergType.asListType().elementType()); - case Collection collection when icebergType != null && icebergType.isListType() -> - convertList(collection, icebergType.asListType().elementType()); - case Record nestedRecord when icebergType != null && icebergType.isStructType() -> - new DelegatedRecord(nestedRecord, icebergType.asStructType()); - case Map map when icebergType != null && icebergType.isMapType() -> - convertMap(map, icebergType.asMapType()); - case null, default -> value; + // Recursively convert complex types against the matching Iceberg type + case null, default -> convertComplexValue(value, icebergType); }; } + /** + * Recursively convert array, collection, nested record, and map values against the matching Iceberg type. + * The value is returned unchanged when the target Iceberg type is unknown or does not describe a complex type + * matching the value. + */ + private static Object convertComplexValue(final Object value, final Type icebergType) { + if (icebergType == null) { + return value; + } + + if (icebergType.isListType()) { + final Type elementType = icebergType.asListType().elementType(); + if (value instanceof Object[] array) { + return convertList(Arrays.asList(array), elementType); + } + if (value instanceof Collection collection) { + return convertList(collection, elementType); + } + } else if (icebergType.isStructType() && value instanceof Record nestedRecord) { + return new DelegatedRecord(nestedRecord, icebergType.asStructType()); + } else if (icebergType.isMapType() && value instanceof Map map) { + return convertMap(map, icebergType.asMapType()); + } + + return value; + } + private static List convertList(final Collection collection, final Type elementType) { final List converted = new ArrayList<>(collection.size()); for (final Object element : collection) { From fc938518a71135a5c4e89a0dbe3445825e148783 Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Fri, 10 Jul 2026 22:26:17 +0200 Subject: [PATCH 3/9] tests: use static constants for field names and values --- .../parquet/ParquetIcebergWriterTest.java | 38 +++++++++---- .../iceberg/record/RecordConverterTest.java | 54 +++++++++++-------- 2 files changed, 61 insertions(+), 31 deletions(-) diff --git a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-parquet-writer/src/test/java/org/apache/nifi/services/iceberg/parquet/ParquetIcebergWriterTest.java b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-parquet-writer/src/test/java/org/apache/nifi/services/iceberg/parquet/ParquetIcebergWriterTest.java index b0159191fe44..1a29da010460 100644 --- a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-parquet-writer/src/test/java/org/apache/nifi/services/iceberg/parquet/ParquetIcebergWriterTest.java +++ b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-parquet-writer/src/test/java/org/apache/nifi/services/iceberg/parquet/ParquetIcebergWriterTest.java @@ -78,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 TAGS_FIELD_VALUE = List.of("a", "b"); + + private static final Map ATTRIBUTES_FIELD_VALUE = Map.of("k", "v"); + private ParquetIcebergWriter parquetIcebergWriter; private TestRunner runner; @@ -201,14 +219,14 @@ void testWriteDataFilesComplexTypes() throws IOException { runner.enableControllerService(parquetIcebergWriter); final Types.StructType nestedStruct = Types.StructType.of( - Types.NestedField.optional(10, "city", Types.StringType.get()) + Types.NestedField.optional(10, CITY_FIELD_NAME, Types.StringType.get()) ); final Schema schema = new Schema( - Types.NestedField.required(1, "id", Types.StringType.get()), - Types.NestedField.optional(2, "tags", + 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", nestedStruct), - Types.NestedField.optional(5, "attributes", + 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(); @@ -219,13 +237,13 @@ void testWriteDataFilesComplexTypes() throws IOException { final IcebergRowWriter rowWriter = parquetIcebergWriter.getRowWriter(table); final GenericRecord address = GenericRecord.create(nestedStruct); - address.setField("city", "Berlin"); + address.setField(CITY_FIELD_NAME, CITY_FIELD_VALUE); final GenericRecord row = GenericRecord.create(schema); - row.setField("id", "row-1"); - row.setField("tags", List.of("a", "b")); - row.setField("address", address); - row.setField("attributes", Map.of("k", "v")); + 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(); diff --git a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/RecordConverterTest.java b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/RecordConverterTest.java index 0b77598b2a09..d20ea2550e08 100644 --- a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/RecordConverterTest.java +++ b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/RecordConverterTest.java @@ -37,6 +37,18 @@ class RecordConverterTest { + private static final String CITY_FIELD_NAME = "city"; + + private static final String CREATED_FIELD_NAME = "created"; + + private static final String NAME_FIELD_NAME = "name"; + + private static final String ITEMS_FIELD_NAME = "items"; + + private static final String CITY_FIELD_VALUE = "Berlin"; + + private static final String NAME_FIELD_VALUE = "widget"; + @Test void testConvertPrimitiveArrayToList() { final Types.ListType listType = Types.ListType.ofOptional(1, Types.StringType.get()); @@ -62,33 +74,33 @@ void testConvertArrayElementDateTime() { @Test void testConvertNestedRecordToStructLike() { final Types.StructType structType = Types.StructType.of( - Types.NestedField.optional(1, "city", Types.StringType.get()) + Types.NestedField.optional(1, CITY_FIELD_NAME, Types.StringType.get()) ); final RecordSchema nestedSchema = new SimpleRecordSchema(List.of( - new RecordField("city", RecordFieldType.STRING.getDataType()) + new RecordField(CITY_FIELD_NAME, RecordFieldType.STRING.getDataType()) )); final Map nestedValues = new LinkedHashMap<>(); - nestedValues.put("city", "Berlin"); + nestedValues.put(CITY_FIELD_NAME, CITY_FIELD_VALUE); final Record nestedRecord = new MapRecord(nestedSchema, nestedValues); final Object converted = RecordConverter.convertValue(nestedRecord, structType); final StructLike struct = assertInstanceOf(StructLike.class, converted); - assertEquals("Berlin", struct.get(0, String.class)); + assertEquals(CITY_FIELD_VALUE, struct.get(0, String.class)); } @Test void testConvertNestedRecordDateTimeField() { final Types.StructType structType = Types.StructType.of( - Types.NestedField.optional(1, "created", Types.TimestampType.withoutZone()) + Types.NestedField.optional(1, CREATED_FIELD_NAME, Types.TimestampType.withoutZone()) ); final RecordSchema nestedSchema = new SimpleRecordSchema(List.of( - new RecordField("created", RecordFieldType.TIMESTAMP.getDataType()) + new RecordField(CREATED_FIELD_NAME, RecordFieldType.TIMESTAMP.getDataType()) )); final Map nestedValues = new LinkedHashMap<>(); - nestedValues.put("created", java.sql.Timestamp.valueOf("2026-01-01 12:30:45")); + nestedValues.put(CREATED_FIELD_NAME, java.sql.Timestamp.valueOf("2026-01-01 12:30:45")); final Record nestedRecord = new MapRecord(nestedSchema, nestedValues); final Object converted = RecordConverter.convertValue(nestedRecord, structType); @@ -103,14 +115,14 @@ void testConvertMapValues() { 1, 2, Types.StringType.get(), Types.StringType.get() ); final Map map = new LinkedHashMap<>(); - map.put("k1", "v1"); - map.put("k2", "v2"); + map.put(CITY_FIELD_NAME, CITY_FIELD_VALUE); + map.put(NAME_FIELD_NAME, NAME_FIELD_VALUE); final Object converted = RecordConverter.convertValue(map, mapType); final Map resultMap = assertInstanceOf(Map.class, converted); - assertEquals("v1", resultMap.get("k1")); - assertEquals("v2", resultMap.get("k2")); + assertEquals(CITY_FIELD_VALUE, resultMap.get(CITY_FIELD_NAME)); + assertEquals(NAME_FIELD_VALUE, resultMap.get(NAME_FIELD_NAME)); } @Test @@ -119,44 +131,44 @@ void testConvertMapDateTimeValue() { 1, 2, Types.StringType.get(), Types.DateType.get() ); final Map map = new LinkedHashMap<>(); - map.put("day", java.sql.Date.valueOf("2026-02-03")); + map.put(CREATED_FIELD_NAME, java.sql.Date.valueOf("2026-02-03")); final Object converted = RecordConverter.convertValue(map, mapType); final Map resultMap = assertInstanceOf(Map.class, converted); - assertEquals(LocalDate.of(2026, 2, 3), resultMap.get("day")); + assertEquals(LocalDate.of(2026, 2, 3), resultMap.get(CREATED_FIELD_NAME)); } @Test void testGetConvertedRecordArrayOfStructs() { final Types.StructType elementStruct = Types.StructType.of( - Types.NestedField.optional(2, "name", Types.StringType.get()) + Types.NestedField.optional(2, NAME_FIELD_NAME, Types.StringType.get()) ); final Types.StructType struct = Types.StructType.of( - Types.NestedField.optional(1, "items", + Types.NestedField.optional(1, ITEMS_FIELD_NAME, Types.ListType.ofOptional(3, elementStruct)) ); final RecordSchema elementSchema = new SimpleRecordSchema(List.of( - new RecordField("name", RecordFieldType.STRING.getDataType()) + new RecordField(NAME_FIELD_NAME, RecordFieldType.STRING.getDataType()) )); final Map elementValues = new LinkedHashMap<>(); - elementValues.put("name", "widget"); + elementValues.put(NAME_FIELD_NAME, NAME_FIELD_VALUE); final Record element = new MapRecord(elementSchema, elementValues); final RecordSchema schema = new SimpleRecordSchema(List.of( - new RecordField("items", + new RecordField(ITEMS_FIELD_NAME, RecordFieldType.ARRAY.getArrayDataType(RecordFieldType.RECORD.getRecordDataType(elementSchema))) )); final Map values = new LinkedHashMap<>(); - values.put("items", new Object[] {element}); + values.put(ITEMS_FIELD_NAME, new Object[] {element}); final Record record = new MapRecord(schema, values); final org.apache.iceberg.data.Record converted = new DelegatedRecord(record, struct); - final Object items = converted.getField("items"); + final Object items = converted.getField(ITEMS_FIELD_NAME); final List list = assertInstanceOf(List.class, items); final StructLike first = assertInstanceOf(StructLike.class, list.get(0)); - assertEquals("widget", first.get(0, String.class)); + assertEquals(NAME_FIELD_VALUE, first.get(0, String.class)); } } From 93c668ba5050d41ef8768d19ddb3535614513249 Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Tue, 28 Jul 2026 07:46:26 +0200 Subject: [PATCH 4/9] fix: pre-allocate LinkedHashMap in convertMap --- .../apache/nifi/processors/iceberg/record/RecordConverter.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java index 4cddb93b4ac0..1af774298da1 100644 --- a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java +++ b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java @@ -120,7 +120,8 @@ private static List convertList(final Collection collection, final Ty } private static Map convertMap(final Map map, final Types.MapType mapType) { - final Map converted = new LinkedHashMap<>(); + // Using LinkedHashMap here to keep input ordering for deterministic flows. + final Map 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()); From 313d966d97c125987aa7ce32ec30381fe47aa13e Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Tue, 28 Jul 2026 08:12:34 +0200 Subject: [PATCH 5/9] fix: add RecordFieldType.CHOICE to required conversion list + regression test --- .../iceberg/record/RecordConverter.java | 4 +- .../iceberg/record/RecordConverterTest.java | 48 +++++++++++++++++++ 2 files changed, 51 insertions(+), 1 deletion(-) diff --git a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java index 1af774298da1..3a69a21344a1 100644 --- a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java +++ b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java @@ -46,7 +46,9 @@ class RecordConverter { RecordFieldType.TIME, RecordFieldType.ARRAY, RecordFieldType.RECORD, - RecordFieldType.MAP + RecordFieldType.MAP, + // CHOICE can wrap any of the above, so it must also trigger conversion. + RecordFieldType.CHOICE ); /** diff --git a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/RecordConverterTest.java b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/RecordConverterTest.java index d20ea2550e08..4cba9d521182 100644 --- a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/RecordConverterTest.java +++ b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/RecordConverterTest.java @@ -45,10 +45,16 @@ class RecordConverterTest { private static final String ITEMS_FIELD_NAME = "items"; + private static final String ADDRESS_FIELD_NAME = "address"; + + private static final String ID_FIELD_NAME = "id"; + private static final String CITY_FIELD_VALUE = "Berlin"; private static final String NAME_FIELD_VALUE = "widget"; + private static final String ID_FIELD_VALUE = "row-1"; + @Test void testConvertPrimitiveArrayToList() { final Types.ListType listType = Types.ListType.ofOptional(1, Types.StringType.get()); @@ -171,4 +177,46 @@ void testGetConvertedRecordArrayOfStructs() { final StructLike first = assertInstanceOf(StructLike.class, list.get(0)); assertEquals(NAME_FIELD_VALUE, first.get(0, String.class)); } + + /** + * A Record field declared as CHOICE, as schema inference produces when a field is an object in some Records and a + * scalar in others, must still be converted. Conversion is driven by the Iceberg type rather than the Record field + * type, so the CHOICE needs no dedicated handling, but it must not short circuit conversion of the whole Record. + */ + @Test + void testGetConvertedRecordChoiceFieldWithScalarSiblings() { + final Types.StructType nestedStruct = Types.StructType.of( + Types.NestedField.optional(2, CITY_FIELD_NAME, Types.StringType.get()) + ); + final Types.StructType struct = Types.StructType.of( + Types.NestedField.optional(1, ADDRESS_FIELD_NAME, nestedStruct), + Types.NestedField.optional(3, ID_FIELD_NAME, Types.StringType.get()) + ); + + final RecordSchema nestedSchema = new SimpleRecordSchema(List.of( + new RecordField(CITY_FIELD_NAME, RecordFieldType.STRING.getDataType()) + )); + final Map nestedValues = new LinkedHashMap<>(); + nestedValues.put(CITY_FIELD_NAME, CITY_FIELD_VALUE); + final Record nestedRecord = new MapRecord(nestedSchema, nestedValues); + + // Every field other than the CHOICE is a scalar, so the CHOICE alone must require conversion + final RecordSchema schema = new SimpleRecordSchema(List.of( + new RecordField(ADDRESS_FIELD_NAME, RecordFieldType.CHOICE.getChoiceDataType( + RecordFieldType.RECORD.getRecordDataType(nestedSchema), + RecordFieldType.STRING.getDataType())), + new RecordField(ID_FIELD_NAME, RecordFieldType.STRING.getDataType()) + )); + final Map values = new LinkedHashMap<>(); + values.put(ADDRESS_FIELD_NAME, nestedRecord); + values.put(ID_FIELD_NAME, ID_FIELD_VALUE); + final Record record = new MapRecord(schema, values); + + final org.apache.iceberg.data.Record converted = new DelegatedRecord(record, struct); + final Object address = converted.getField(ADDRESS_FIELD_NAME); + + final StructLike addressStruct = assertInstanceOf(StructLike.class, address); + assertEquals(CITY_FIELD_VALUE, addressStruct.get(0, String.class)); + assertEquals(ID_FIELD_VALUE, converted.getField(ID_FIELD_NAME)); + } } From c3be398ad1b3adef3909390a2deaf269f817484a Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Thu, 27 Aug 2026 07:36:44 +0200 Subject: [PATCH 6/9] refactor: single return in getConvertedRecord --- .../iceberg/record/RecordConverter.java | 25 +++++++++++-------- 1 file changed, 14 insertions(+), 11 deletions(-) diff --git a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java index 3a69a21344a1..a63f56705a49 100644 --- a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java +++ b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java @@ -59,20 +59,23 @@ class RecordConverter { * @return Input Record or new Record with converted field values */ static Record getConvertedRecord(final Record inputRecord, final Types.StructType struct) { - final RecordSchema recordSchema = inputRecord.getSchema(); - if (!isConversionRequired(recordSchema)) { - return inputRecord; - } + final Record convertedRecord; - final Map values = inputRecord.toMap(); - final Map convertedValues = new LinkedHashMap<>(); - for (final Map.Entry entry : values.entrySet()) { - final String field = entry.getKey(); - final Type fieldType = fieldType(struct, field); - convertedValues.put(field, convertValue(entry.getValue(), fieldType)); + final RecordSchema recordSchema = inputRecord.getSchema(); + if (isConversionRequired(recordSchema)) { + final Map values = inputRecord.toMap(); + final Map convertedValues = new LinkedHashMap<>(values.size()); + for (final Map.Entry 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 new MapRecord(recordSchema, convertedValues); + return convertedRecord; } static Object convertValue(final Object value, final Type icebergType) { From 10dac695d1d21a0a9a1bd4b5533449b00ad665d3 Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Thu, 27 Aug 2026 07:39:53 +0200 Subject: [PATCH 7/9] refactor: single return in convertComplexValue --- .../iceberg/record/RecordConverter.java | 50 ++++++++++++------- 1 file changed, 33 insertions(+), 17 deletions(-) diff --git a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java index a63f56705a49..6fb337603132 100644 --- a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java +++ b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java @@ -90,30 +90,46 @@ static Object convertValue(final Object value, final Type icebergType) { } /** - * Recursively convert array, collection, nested record, and map values against the matching Iceberg type. - * The value is returned unchanged when the target Iceberg type is unknown or does not describe a complex type - * matching the value. + * 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) { - if (icebergType == null) { - return value; - } + final Object convertedValue; - if (icebergType.isListType()) { - final Type elementType = icebergType.asListType().elementType(); - if (value instanceof Object[] array) { - return convertList(Arrays.asList(array), elementType); - } - if (value instanceof Collection collection) { - return convertList(collection, elementType); - } + if (icebergType == null) { + convertedValue = value; + } else if (icebergType.isListType()) { + convertedValue = convertListValue(value, icebergType.asListType()); } else if (icebergType.isStructType() && value instanceof Record nestedRecord) { - return new DelegatedRecord(nestedRecord, icebergType.asStructType()); + convertedValue = new DelegatedRecord(nestedRecord, icebergType.asStructType()); } else if (icebergType.isMapType() && value instanceof Map map) { - return convertMap(map, icebergType.asMapType()); + convertedValue = convertMap(map, icebergType.asMapType()); + } else { + convertedValue = value; } - return value; + return convertedValue; + } + + /** + * 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) { + case Object[] array -> convertList(Arrays.asList(array), elementType); + case Collection collection -> convertList(collection, elementType); + case null, default -> value; + }; } private static List convertList(final Collection collection, final Type elementType) { From 218b4866c23e8085e163180561f36e38aa30d9c7 Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Thu, 27 Aug 2026 07:44:20 +0200 Subject: [PATCH 8/9] refactor: single return in fieldType and isConversionRequired Collapse the remaining short-circuit returns in RecordConverter for consistency. --- .../iceberg/record/RecordConverter.java | 17 ++++++----------- 1 file changed, 6 insertions(+), 11 deletions(-) diff --git a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java index 6fb337603132..79fc69c43f51 100644 --- a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java +++ b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java @@ -18,6 +18,7 @@ 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; import org.apache.nifi.serialization.record.RecordField; @@ -152,20 +153,14 @@ private static Map convertMap(final Map map, final Types.M } private static Type fieldType(final Types.StructType struct, final String fieldName) { - if (struct == null) { - return null; - } - final Types.NestedField nestedField = struct.field(fieldName); + final Types.NestedField nestedField = struct == null ? null : struct.field(fieldName); return nestedField == null ? null : nestedField.type(); } private static boolean isConversionRequired(final RecordSchema recordSchema) { - for (final RecordField field : recordSchema.getFields()) { - final RecordFieldType recordFieldType = field.getDataType().getFieldType(); - if (CONVERSION_REQUIRED_FIELD_TYPES.contains(recordFieldType)) { - return true; - } - } - return false; + return recordSchema.getFields().stream() + .map(RecordField::getDataType) + .map(DataType::getFieldType) + .anyMatch(CONVERSION_REQUIRED_FIELD_TYPES::contains); } } From 270dc35ba3b33a3da6e4fe684986b1c9952a1a67 Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Thu, 27 Aug 2026 07:52:40 +0200 Subject: [PATCH 9/9] feat: convert Timestamp to OffsetDateTime for Iceberg types adjusted to UTC Iceberg timestamptz and timestamptz_ns columns require an OffsetDateTime, so resolve the conversion from the target Iceberg type instead of always producing a LocalDateTime. A Timestamp identifies an instant, so the adjusted conversion preserves that instant expressed at UTC. --- .../iceberg/record/RecordConverter.java | 31 ++++++- .../iceberg/record/RecordConverterTest.java | 81 ++++++++++++++++++- 2 files changed, 110 insertions(+), 2 deletions(-) diff --git a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java index 79fc69c43f51..27dad53acb45 100644 --- a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java +++ b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java @@ -28,6 +28,7 @@ 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; @@ -82,7 +83,7 @@ static Record getConvertedRecord(final Record inputRecord, final Types.StructTyp 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 -> timestamp.toLocalDateTime(); + 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 @@ -90,6 +91,34 @@ static Object convertValue(final Object value, final Type icebergType) { }; } + /** + * 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 * diff --git a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/RecordConverterTest.java b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/RecordConverterTest.java index 4cba9d521182..3dcc0fcbf2f2 100644 --- a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/RecordConverterTest.java +++ b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/RecordConverterTest.java @@ -17,6 +17,7 @@ package org.apache.nifi.processors.iceberg.record; import org.apache.iceberg.StructLike; +import org.apache.iceberg.types.Type; import org.apache.iceberg.types.Types; import org.apache.nifi.serialization.SimpleRecordSchema; import org.apache.nifi.serialization.record.MapRecord; @@ -25,12 +26,19 @@ import org.apache.nifi.serialization.record.RecordFieldType; import org.apache.nifi.serialization.record.RecordSchema; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; +import java.sql.Timestamp; import java.time.LocalDate; import java.time.LocalDateTime; +import java.time.OffsetDateTime; +import java.time.ZoneOffset; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.stream.Stream; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertInstanceOf; @@ -55,6 +63,8 @@ class RecordConverterTest { private static final String ID_FIELD_VALUE = "row-1"; + private static final LocalDateTime CREATED_LOCAL_DATE_TIME = LocalDateTime.of(2026, 1, 1, 12, 30, 45); + @Test void testConvertPrimitiveArrayToList() { final Types.ListType listType = Types.ListType.ofOptional(1, Types.StringType.get()); @@ -106,7 +116,7 @@ void testConvertNestedRecordDateTimeField() { new RecordField(CREATED_FIELD_NAME, RecordFieldType.TIMESTAMP.getDataType()) )); final Map nestedValues = new LinkedHashMap<>(); - nestedValues.put(CREATED_FIELD_NAME, java.sql.Timestamp.valueOf("2026-01-01 12:30:45")); + nestedValues.put(CREATED_FIELD_NAME, Timestamp.valueOf("2026-01-01 12:30:45")); final Record nestedRecord = new MapRecord(nestedSchema, nestedValues); final Object converted = RecordConverter.convertValue(nestedRecord, structType); @@ -115,6 +125,75 @@ void testConvertNestedRecordDateTimeField() { assertEquals(LocalDateTime.of(2026, 1, 1, 12, 30, 45), struct.get(0, LocalDateTime.class)); } + /** + * Iceberg Types not adjusted to UTC require a LocalDateTime. The Iceberg Type is not resolved for every field, + * so an unknown Type must retain the same conversion. + */ + @ParameterizedTest + @MethodSource + void testConvertTimestampNotAdjustedToUtc(final Type icebergType) { + final Timestamp timestamp = Timestamp.valueOf(CREATED_LOCAL_DATE_TIME); + + final Object converted = RecordConverter.convertValue(timestamp, icebergType); + + assertEquals(CREATED_LOCAL_DATE_TIME, converted); + } + + private static Stream testConvertTimestampNotAdjustedToUtc() { + return Stream.of( + Arguments.of(Types.TimestampType.withoutZone()), + Arguments.of(Types.TimestampNanoType.withoutZone()), + Arguments.of((Type) null) + ); + } + + /** + * Iceberg timestamptz columns require an OffsetDateTime rather than a LocalDateTime. A Timestamp identifies an + * instant, so the converted value must describe that same instant expressed at UTC. + */ + @ParameterizedTest + @MethodSource + void testConvertTimestampAdjustedToUtc(final Type icebergType) { + final Timestamp timestamp = Timestamp.valueOf(CREATED_LOCAL_DATE_TIME); + + final Object converted = RecordConverter.convertValue(timestamp, icebergType); + + final OffsetDateTime offsetDateTime = assertInstanceOf(OffsetDateTime.class, converted); + assertEquals(ZoneOffset.UTC, offsetDateTime.getOffset()); + assertEquals(timestamp.toInstant(), offsetDateTime.toInstant()); + } + + private static Stream testConvertTimestampAdjustedToUtc() { + return Stream.of( + Arguments.of(Types.TimestampType.withZone()), + Arguments.of(Types.TimestampNanoType.withZone()) + ); + } + + /** + * A timestamptz column nested inside a struct must be converted through the recursive path, which requires the + * Iceberg Type of the nested field to be resolved and passed down. + */ + @Test + void testGetConvertedRecordNestedTimestampWithZone() { + final Types.StructType structType = Types.StructType.of( + Types.NestedField.optional(1, CREATED_FIELD_NAME, Types.TimestampType.withZone()) + ); + + final RecordSchema nestedSchema = new SimpleRecordSchema(List.of( + new RecordField(CREATED_FIELD_NAME, RecordFieldType.TIMESTAMP.getDataType()) + )); + final Timestamp timestamp = Timestamp.valueOf(CREATED_LOCAL_DATE_TIME); + final Map nestedValues = new LinkedHashMap<>(); + nestedValues.put(CREATED_FIELD_NAME, timestamp); + final Record nestedRecord = new MapRecord(nestedSchema, nestedValues); + + final Object converted = RecordConverter.convertValue(nestedRecord, structType); + + final StructLike struct = assertInstanceOf(StructLike.class, converted); + assertEquals(timestamp.toInstant(), struct.get(0, OffsetDateTime.class).toInstant()); + } + @Test void testConvertMapValues() { final Types.MapType mapType = Types.MapType.ofOptional(