maltesander commented on code in PR #11391:
URL: https://github.com/apache/nifi/pull/11391#discussion_r3563489879
##########
nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java:
##########
@@ -39,63 +43,107 @@ 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
);
/**
- * 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<String, Object> 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<String, Object> values) {
+ 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);
+ 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 null, default -> value;
+ // Recursively convert complex types against the matching Iceberg
type
+ case null, default -> convertComplexValue(value, icebergType);
};
}
- private static boolean isConversionRequired(final RecordSchema
recordSchema) {
- final List<RecordField> fields = recordSchema.getFields();
+ /**
+ * 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());
+ }
- for (final RecordField field : fields) {
- final DataType dataType = field.getDataType();
- final RecordFieldType recordFieldType = dataType.getFieldType();
+ return value;
+ }
+
+ 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;
+ }
+
+ private static Map<Object, Object> convertMap(final Map<?, ?> map, final
Types.MapType mapType) {
+ final Map<Object, Object> converted = new LinkedHashMap<>();
Review Comment:
LinkedHashMap vs HashMap might be debateable?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
