NIFI-16069 - PutIcebergRecord fails with ClassCastException when writing complex types (arrays, maps, nested records)#11391
NIFI-16069 - PutIcebergRecord fails with ClassCastException when writing complex types (arrays, maps, nested records)#11391maltesander wants to merge 3 commits into
Conversation
…ing complex types (arrays, maps, nested records)
exceptionfactory
left a comment
exceptionfactory
left a comment
There was a problem hiding this comment.
Thanks for proposing this improvement @maltesander. The initial version of Iceberg integration did not include supported for nested and complex types, so this is an important area of improvement. The general approach looks good, and the tests are helpful. I plan on taking a closer look at the conversion details, I noted a few minor recommendations for now.
| 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()); |
There was a problem hiding this comment.
The nested conditionals are difficult to read, I recommend breaking this out in more detail to make it easier to follow
There was a problem hiding this comment.
Thank you for the feedback @exceptionfactory. I tried to simplify here d174fd3.
I introduced a convertComplexValue method to handle this.
| 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())) | ||
| ); |
There was a problem hiding this comment.
The field names should be declared statically and reused in both the input Schema and output row.
| } | ||
|
|
||
| private static Map<Object, Object> convertMap(final Map<?, ?> map, final Types.MapType mapType) { | ||
| final Map<Object, Object> converted = new LinkedHashMap<>(); |
There was a problem hiding this comment.
LinkedHashMap vs HashMap might be debateable?
2 participants
Summary
NIFI-16069 - PutIcebergRecord fails with ClassCastException when writing complex types (arrays, maps, nested records)
PutIcebergRecord fails to write FlowFiles whose schema contains complex/nested types like Iceberg list, map, or struct columns. RecordConverter only translates top-level scalar values (java.sql timestamp/date/time -> java.time) and passes complex values through unchanged.
As a result, values reach Iceberg's Parquet writer in NiFi's native representation, which is incompatible with what Iceberg expects:
Because conversion is gated on scalar field types only, records consisting solely of complex fields skip conversion entirely.
Tracking
Please complete the following tracking steps prior to pull request creation.
Issue Tracking
Pull Request Tracking
NIFI-00000NIFI-00000VerifiedstatusPull Request Formatting
mainbranchVerification
Please indicate the verification steps performed prior to pull request creation.
Build
./mvnw clean install -P contrib-checkLicensing
LICENSEandNOTICEfilesDocumentation