exceptionfactory commented on code in PR #11391:
URL: https://github.com/apache/nifi/pull/11391#discussion_r3561527508
##########
nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-parquet-writer/src/test/java/org/apache/nifi/services/iceberg/parquet/ParquetIcebergWriterTest.java:
##########
@@ -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()))
+ );
Review Comment:
The field names should be declared statically and reused in both the input
Schema and output row.
##########
nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java:
##########
@@ -39,63 +43,87 @@ 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 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());
Review Comment:
The nested conditionals are difficult to read, I recommend breaking this out
in more detail to make it easier to follow
--
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]
