maltesander commented on code in PR #11391:
URL: https://github.com/apache/nifi/pull/11391#discussion_r3869127809
##########
nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java:
##########
@@ -39,63 +43,110 @@ 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,
+ // CHOICE can wrap any of the above, so it must also trigger
conversion.
+ RecordFieldType.CHOICE
);
/**
- * 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) {
Review Comment:
fixed
https://github.com/apache/nifi/pull/11391/changes/10dac695d1d21a0a9a1bd4b5533449b00ad665d3
##########
nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java:
##########
@@ -39,63 +43,110 @@ 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,
+ // CHOICE can wrap any of the above, so it must also trigger
conversion.
+ RecordFieldType.CHOICE
);
/**
- * 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.
+ */
Review Comment:
fixed
https://github.com/apache/nifi/pull/11391/changes/10dac695d1d21a0a9a1bd4b5533449b00ad665d3
--
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]
