Skip to content

Commit eb7a57f

Browse files
maltesandermaltesander
committed
fix: patch PutIcebergRecord processor for nested types and use field access per name and not id
1 parent 46ad345 commit eb7a57f

2 files changed

Lines changed: 578 additions & 0 deletions

File tree

‎nifi/stackable/patches/2.9.0/0008-NIFI-16069-PutIcebergRecord-fails-with-ClassCastExce.patch‎

Lines changed: 390 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,390 @@
1+
From be1604e5d6eedbf8d35477f826c9567973e4bc56 Mon Sep 17 00:00:00 2001
2+
From: Malte Sander <malte.sander.it@gmail.com>
3+
Date: Thu, 2 Jul 2026 20:52:20 +0200
4+
Subject: NIFI-16069 - PutIcebergRecord fails with ClassCastException when
5+
writing complex types (arrays, maps, nested records)
6+
7+
---
8+
.../parquet/ParquetIcebergWriterTest.java | 39 +++++
9+
.../iceberg/record/DelegatedRecord.java | 2 +-
10+
.../iceberg/record/RecordConverter.java | 80 ++++++---
11+
.../iceberg/record/RecordConverterTest.java | 162 ++++++++++++++++++
12+
4 files changed, 256 insertions(+), 27 deletions(-)
13+
create mode 100644 nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/RecordConverterTest.java
14+
15+
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
16+
index 636b0ba091..b0159191fe 100644
17+
--- a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-parquet-writer/src/test/java/org/apache/nifi/services/iceberg/parquet/ParquetIcebergWriterTest.java
18+
+++ b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-parquet-writer/src/test/java/org/apache/nifi/services/iceberg/parquet/ParquetIcebergWriterTest.java
19+
@@ -47,6 +47,8 @@ import java.io.IOException;
20+
import java.time.LocalDate;
21+
import java.time.LocalDateTime;
22+
import java.time.LocalTime;
23+
+import java.util.List;
24+
+import java.util.Map;
25+
26+
import static org.junit.jupiter.api.Assertions.assertEquals;
27+
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
28+
@@ -194,6 +196,43 @@ class ParquetIcebergWriterTest {
29+
assertEquals(microsecondsExpected, partitionField);
30+
}
31+
32+
+ @Test
33+
+ void testWriteDataFilesComplexTypes() throws IOException {
34+
+ runner.enableControllerService(parquetIcebergWriter);
35+
+
36+
+ final Types.StructType nestedStruct = Types.StructType.of(
37+
+ Types.NestedField.optional(10, "city", Types.StringType.get())
38+
+ );
39+
+ final Schema schema = new Schema(
40+
+ Types.NestedField.required(1, "id", Types.StringType.get()),
41+
+ Types.NestedField.optional(2, "tags",
42+
+ Types.ListType.ofOptional(3, Types.StringType.get())),
43+
+ Types.NestedField.optional(4, "address", nestedStruct),
44+
+ Types.NestedField.optional(5, "attributes",
45+
+ Types.MapType.ofOptional(6, 7, Types.StringType.get(), Types.StringType.get()))
46+
+ );
47+
+ final InMemoryOutputFile outputFile = new InMemoryOutputFile();
48+
+ final PartitionSpec partitionSpec = PartitionSpec.unpartitioned();
49+
+ setTable(schema, partitionSpec, outputFile);
50+
+ when(locationProvider.newDataLocation(anyString())).thenReturn(LOCATION);
51+
+
52+
+ final IcebergRowWriter rowWriter = parquetIcebergWriter.getRowWriter(table);
53+
+
54+
+ final GenericRecord address = GenericRecord.create(nestedStruct);
55+
+ address.setField("city", "Berlin");
56+
+
57+
+ final GenericRecord row = GenericRecord.create(schema);
58+
+ row.setField("id", "row-1");
59+
+ row.setField("tags", List.of("a", "b"));
60+
+ row.setField("address", address);
61+
+ row.setField("attributes", Map.of("k", "v"));
62+
+ rowWriter.write(row);
63+
+
64+
+ final DataFile[] dataFiles = rowWriter.dataFiles();
65+
+ final byte[] serialized = outputFile.toByteArray();
66+
+ assertDataFilesFound(dataFiles, serialized);
67+
+ }
68+
+
69+
private void writeRow(final Schema schema, final IcebergRowWriter rowWriter) throws IOException {
70+
final GenericRecord row = GenericRecord.create(schema);
71+
row.setField(FIRST_FIELD_NAME, FIRST_FIELD_VALUE);
72+
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
73+
index 3267cc3221..ac98c06f37 100644
74+
--- a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/DelegatedRecord.java
75+
+++ b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/DelegatedRecord.java
76+
@@ -38,8 +38,8 @@ public class DelegatedRecord implements Record {
77+
final org.apache.nifi.serialization.record.Record record,
78+
final Types.StructType struct
79+
) {
80+
- this.record = RecordConverter.getConvertedRecord(Objects.requireNonNull(record));
81+
this.struct = Objects.requireNonNull(struct);
82+
+ this.record = RecordConverter.getConvertedRecord(Objects.requireNonNull(record), struct);
83+
}
84+
85+
@Override
86+
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
87+
index b84159d4af..b2212341c8 100644
88+
--- a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java
89+
+++ b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java
90+
@@ -16,7 +16,8 @@
91+
*/
92+
package org.apache.nifi.processors.iceberg.record;
93+
94+
-import org.apache.nifi.serialization.record.DataType;
95+
+import org.apache.iceberg.types.Type;
96+
+import org.apache.iceberg.types.Types;
97+
import org.apache.nifi.serialization.record.MapRecord;
98+
import org.apache.nifi.serialization.record.Record;
99+
import org.apache.nifi.serialization.record.RecordField;
100+
@@ -26,6 +27,9 @@ import org.apache.nifi.serialization.record.RecordSchema;
101+
import java.sql.Date;
102+
import java.sql.Time;
103+
import java.sql.Timestamp;
104+
+import java.util.ArrayList;
105+
+import java.util.Arrays;
106+
+import java.util.Collection;
107+
import java.util.LinkedHashMap;
108+
import java.util.List;
109+
import java.util.Map;
110+
@@ -39,63 +43,87 @@ class RecordConverter {
111+
private static final Set<RecordFieldType> CONVERSION_REQUIRED_FIELD_TYPES = Set.of(
112+
RecordFieldType.TIMESTAMP,
113+
RecordFieldType.DATE,
114+
- RecordFieldType.TIME
115+
+ RecordFieldType.TIME,
116+
+ RecordFieldType.ARRAY,
117+
+ RecordFieldType.RECORD,
118+
+ RecordFieldType.MAP
119+
);
120+
121+
/**
122+
- * Get Converted Record with conditional handling for field values requiring translation
123+
+ * Get Converted Record with recursive, schema-aware handling for field values requiring translation
124+
*
125+
* @param inputRecord Input Record to be converted
126+
+ * @param struct Iceberg Struct Type describing the target field types (may be null for scalar-only conversion)
127+
* @return Input Record or new Record with converted field values
128+
*/
129+
- static Record getConvertedRecord(final Record inputRecord) {
130+
- final Record convertedRecord;
131+
-
132+
+ static Record getConvertedRecord(final Record inputRecord, final Types.StructType struct) {
133+
final RecordSchema recordSchema = inputRecord.getSchema();
134+
- if (isConversionRequired(recordSchema)) {
135+
- final Map<String, Object> values = inputRecord.toMap();
136+
- convertedRecord = getConvertedRecord(recordSchema, values);
137+
- } else {
138+
- convertedRecord = inputRecord;
139+
+ if (!isConversionRequired(recordSchema)) {
140+
+ return inputRecord;
141+
}
142+
143+
- return convertedRecord;
144+
- }
145+
-
146+
- private static Record getConvertedRecord(final RecordSchema recordSchema, final Map<String, Object> values) {
147+
+ final Map<String, Object> values = inputRecord.toMap();
148+
final Map<String, Object> convertedValues = new LinkedHashMap<>();
149+
-
150+
for (final Map.Entry<String, Object> entry : values.entrySet()) {
151+
final String field = entry.getKey();
152+
- final Object value = entry.getValue();
153+
- final Object converted = getConvertedValue(value);
154+
- convertedValues.put(field, converted);
155+
+ final Type fieldType = fieldType(struct, field);
156+
+ convertedValues.put(field, convertValue(entry.getValue(), fieldType));
157+
}
158+
159+
return new MapRecord(recordSchema, convertedValues);
160+
}
161+
162+
- private static Object getConvertedValue(final Object value) {
163+
+ static Object convertValue(final Object value, final Type icebergType) {
164+
return switch (value) {
165+
// Convert java.sql types to corresponding java.time types for Apache Iceberg
166+
case Timestamp timestamp -> timestamp.toLocalDateTime();
167+
case Date date -> date.toLocalDate();
168+
case Time time -> time.toLocalTime();
169+
+ case Object[] array when icebergType != null && icebergType.isListType() ->
170+
+ convertList(Arrays.asList(array), icebergType.asListType().elementType());
171+
+ case Collection<?> collection when icebergType != null && icebergType.isListType() ->
172+
+ convertList(collection, icebergType.asListType().elementType());
173+
+ case Record nestedRecord when icebergType != null && icebergType.isStructType() ->
174+
+ new DelegatedRecord(nestedRecord, icebergType.asStructType());
175+
+ case Map<?, ?> map when icebergType != null && icebergType.isMapType() ->
176+
+ convertMap(map, icebergType.asMapType());
177+
case null, default -> value;
178+
};
179+
}
180+
181+
- private static boolean isConversionRequired(final RecordSchema recordSchema) {
182+
- final List<RecordField> fields = recordSchema.getFields();
183+
+ private static List<Object> convertList(final Collection<?> collection, final Type elementType) {
184+
+ final List<Object> converted = new ArrayList<>(collection.size());
185+
+ for (final Object element : collection) {
186+
+ converted.add(convertValue(element, elementType));
187+
+ }
188+
+ return converted;
189+
+ }
190+
191+
- for (final RecordField field : fields) {
192+
- final DataType dataType = field.getDataType();
193+
- final RecordFieldType recordFieldType = dataType.getFieldType();
194+
+ private static Map<Object, Object> convertMap(final Map<?, ?> map, final Types.MapType mapType) {
195+
+ final Map<Object, Object> converted = new LinkedHashMap<>();
196+
+ for (final Map.Entry<?, ?> entry : map.entrySet()) {
197+
+ final Object key = convertValue(entry.getKey(), mapType.keyType());
198+
+ final Object mappedValue = convertValue(entry.getValue(), mapType.valueType());
199+
+ converted.put(key, mappedValue);
200+
+ }
201+
+ return converted;
202+
+ }
203+
+
204+
+ private static Type fieldType(final Types.StructType struct, final String fieldName) {
205+
+ if (struct == null) {
206+
+ return null;
207+
+ }
208+
+ final Types.NestedField nestedField = struct.field(fieldName);
209+
+ return nestedField == null ? null : nestedField.type();
210+
+ }
211+
+
212+
+ private static boolean isConversionRequired(final RecordSchema recordSchema) {
213+
+ for (final RecordField field : recordSchema.getFields()) {
214+
+ final RecordFieldType recordFieldType = field.getDataType().getFieldType();
215+
if (CONVERSION_REQUIRED_FIELD_TYPES.contains(recordFieldType)) {
216+
return true;
217+
}
218+
}
219+
-
220+
return false;
221+
}
222+
}
223+
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
224+
new file mode 100644
225+
index 0000000000..0b77598b2a
226+
--- /dev/null
227+
+++ b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/RecordConverterTest.java
228+
@@ -0,0 +1,162 @@
229+
+/*
230+
+ * Licensed to the Apache Software Foundation (ASF) under one or more
231+
+ * contributor license agreements. See the NOTICE file distributed with
232+
+ * this work for additional information regarding copyright ownership.
233+
+ * The ASF licenses this file to You under the Apache License, Version 2.0
234+
+ * (the "License"); you may not use this file except in compliance with
235+
+ * the License. You may obtain a copy of the License at
236+
+ *
237+
+ * http://www.apache.org/licenses/LICENSE-2.0
238+
+ *
239+
+ * Unless required by applicable law or agreed to in writing, software
240+
+ * distributed under the License is distributed on an "AS IS" BASIS,
241+
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
242+
+ * See the License for the specific language governing permissions and
243+
+ * limitations under the License.
244+
+ */
245+
+package org.apache.nifi.processors.iceberg.record;
246+
+
247+
+import org.apache.iceberg.StructLike;
248+
+import org.apache.iceberg.types.Types;
249+
+import org.apache.nifi.serialization.SimpleRecordSchema;
250+
+import org.apache.nifi.serialization.record.MapRecord;
251+
+import org.apache.nifi.serialization.record.Record;
252+
+import org.apache.nifi.serialization.record.RecordField;
253+
+import org.apache.nifi.serialization.record.RecordFieldType;
254+
+import org.apache.nifi.serialization.record.RecordSchema;
255+
+import org.junit.jupiter.api.Test;
256+
+
257+
+import java.time.LocalDate;
258+
+import java.time.LocalDateTime;
259+
+import java.util.LinkedHashMap;
260+
+import java.util.List;
261+
+import java.util.Map;
262+
+
263+
+import static org.junit.jupiter.api.Assertions.assertEquals;
264+
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
265+
+
266+
+class RecordConverterTest {
267+
+
268+
+ @Test
269+
+ void testConvertPrimitiveArrayToList() {
270+
+ final Types.ListType listType = Types.ListType.ofOptional(1, Types.StringType.get());
271+
+ final Object[] array = new Object[] {"a", "b", "c"};
272+
+
273+
+ final Object converted = RecordConverter.convertValue(array, listType);
274+
+
275+
+ final List<?> list = assertInstanceOf(List.class, converted);
276+
+ assertEquals(List.of("a", "b", "c"), list);
277+
+ }
278+
+
279+
+ @Test
280+
+ void testConvertArrayElementDateTime() {
281+
+ final Types.ListType listType = Types.ListType.ofOptional(1, Types.DateType.get());
282+
+ final Object[] array = new Object[] {java.sql.Date.valueOf("2026-02-03")};
283+
+
284+
+ final Object converted = RecordConverter.convertValue(array, listType);
285+
+
286+
+ final List<?> list = assertInstanceOf(List.class, converted);
287+
+ assertEquals(List.of(LocalDate.of(2026, 2, 3)), list);
288+
+ }
289+
+
290+
+ @Test
291+
+ void testConvertNestedRecordToStructLike() {
292+
+ final Types.StructType structType = Types.StructType.of(
293+
+ Types.NestedField.optional(1, "city", Types.StringType.get())
294+
+ );
295+
+
296+
+ final RecordSchema nestedSchema = new SimpleRecordSchema(List.of(
297+
+ new RecordField("city", RecordFieldType.STRING.getDataType())
298+
+ ));
299+
+ final Map<String, Object> nestedValues = new LinkedHashMap<>();
300+
+ nestedValues.put("city", "Berlin");
301+
+ final Record nestedRecord = new MapRecord(nestedSchema, nestedValues);
302+
+
303+
+ final Object converted = RecordConverter.convertValue(nestedRecord, structType);
304+
+
305+
+ final StructLike struct = assertInstanceOf(StructLike.class, converted);
306+
+ assertEquals("Berlin", struct.get(0, String.class));
307+
+ }
308+
+
309+
+ @Test
310+
+ void testConvertNestedRecordDateTimeField() {
311+
+ final Types.StructType structType = Types.StructType.of(
312+
+ Types.NestedField.optional(1, "created", Types.TimestampType.withoutZone())
313+
+ );
314+
+
315+
+ final RecordSchema nestedSchema = new SimpleRecordSchema(List.of(
316+
+ new RecordField("created", RecordFieldType.TIMESTAMP.getDataType())
317+
+ ));
318+
+ final Map<String, Object> nestedValues = new LinkedHashMap<>();
319+
+ nestedValues.put("created", java.sql.Timestamp.valueOf("2026-01-01 12:30:45"));
320+
+ final Record nestedRecord = new MapRecord(nestedSchema, nestedValues);
321+
+
322+
+ final Object converted = RecordConverter.convertValue(nestedRecord, structType);
323+
+
324+
+ final StructLike struct = assertInstanceOf(StructLike.class, converted);
325+
+ assertEquals(LocalDateTime.of(2026, 1, 1, 12, 30, 45), struct.get(0, LocalDateTime.class));
326+
+ }
327+
+
328+
+ @Test
329+
+ void testConvertMapValues() {
330+
+ final Types.MapType mapType = Types.MapType.ofOptional(
331+
+ 1, 2, Types.StringType.get(), Types.StringType.get()
332+
+ );
333+
+ final Map<String, Object> map = new LinkedHashMap<>();
334+
+ map.put("k1", "v1");
335+
+ map.put("k2", "v2");
336+
+
337+
+ final Object converted = RecordConverter.convertValue(map, mapType);
338+
+
339+
+ final Map<?, ?> resultMap = assertInstanceOf(Map.class, converted);
340+
+ assertEquals("v1", resultMap.get("k1"));
341+
+ assertEquals("v2", resultMap.get("k2"));
342+
+ }
343+
+
344+
+ @Test
345+
+ void testConvertMapDateTimeValue() {
346+
+ final Types.MapType mapType = Types.MapType.ofOptional(
347+
+ 1, 2, Types.StringType.get(), Types.DateType.get()
348+
+ );
349+
+ final Map<String, Object> map = new LinkedHashMap<>();
350+
+ map.put("day", java.sql.Date.valueOf("2026-02-03"));
351+
+
352+
+ final Object converted = RecordConverter.convertValue(map, mapType);
353+
+
354+
+ final Map<?, ?> resultMap = assertInstanceOf(Map.class, converted);
355+
+ assertEquals(LocalDate.of(2026, 2, 3), resultMap.get("day"));
356+
+ }
357+
+
358+
+ @Test
359+
+ void testGetConvertedRecordArrayOfStructs() {
360+
+ final Types.StructType elementStruct = Types.StructType.of(
361+
+ Types.NestedField.optional(2, "name", Types.StringType.get())
362+
+ );
363+
+ final Types.StructType struct = Types.StructType.of(
364+
+ Types.NestedField.optional(1, "items",
365+
+ Types.ListType.ofOptional(3, elementStruct))
366+
+ );
367+
+
368+
+ final RecordSchema elementSchema = new SimpleRecordSchema(List.of(
369+
+ new RecordField("name", RecordFieldType.STRING.getDataType())
370+
+ ));
371+
+ final Map<String, Object> elementValues = new LinkedHashMap<>();
372+
+ elementValues.put("name", "widget");
373+
+ final Record element = new MapRecord(elementSchema, elementValues);
374+
+
375+
+ final RecordSchema schema = new SimpleRecordSchema(List.of(
376+
+ new RecordField("items",
377+
+ RecordFieldType.ARRAY.getArrayDataType(RecordFieldType.RECORD.getRecordDataType(elementSchema)))
378+
+ ));
379+
+ final Map<String, Object> values = new LinkedHashMap<>();
380+
+ values.put("items", new Object[] {element});
381+
+ final Record record = new MapRecord(schema, values);
382+
+
383+
+ final org.apache.iceberg.data.Record converted = new DelegatedRecord(record, struct);
384+
+ final Object items = converted.getField("items");
385+
+
386+
+ final List<?> list = assertInstanceOf(List.class, items);
387+
+ final StructLike first = assertInstanceOf(StructLike.class, list.get(0));
388+
+ assertEquals("widget", first.get(0, String.class));
389+
+ }
390+
+}

0 commit comments

Comments
 (0)