Skip to content

NIFI-16074 - PutIcebergRecord missing convertors #11423

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
NirYanay2005 wants to merge 3 commits into
base: main
Choose a base branch
from
Open

NIFI-16074 - PutIcebergRecord missing convertors #11423

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
4021492
NIFI-16074 Added conversions for more Iceberg types
NirYanay2005 Jul 13, 2026
6e9d30e
NIFI-16074 Fix DelegatedRecord positional get/set to resolve by Icebe…
NirYanay2005 Jul 13, 2026
a9f2bb9
Added variant type support
NirYanay2005 Jul 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 5 additions & 6 deletions ...g-processors/src/main/java/org/apache/nifi/processors/iceberg/record/DelegatedRecord.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@
import org.apache.iceberg.data.Record;
import org.apache.iceberg.types.Types;
import org.apache.nifi.serialization.record.MapRecord;
import org.apache.nifi.serialization.record.RecordField;

import java.util.Collections;
import java.util.LinkedHashMap;
Expand All @@ -38,7 +37,7 @@ public DelegatedRecord(
final org.apache.nifi.serialization.record.Record record,
final Types.StructType struct
) {
this.record = RecordConverter.getConvertedRecord(Objects.requireNonNull(record));
this.record = RecordConverter.getConvertedRecord(Objects.requireNonNull(record), struct);
this.struct = Objects.requireNonNull(struct);
}

Expand Down Expand Up @@ -77,8 +76,8 @@ public void setField(final String fieldName, final Object fieldValue) {
*/
@Override
public Object get(final int position) {
final RecordField recordField = record.getSchema().getField(position);
return record.getValue(recordField);
final String fieldName = struct.fields().get(position).name();
return record.getValue(fieldName);
}

/**
Expand Down Expand Up @@ -141,8 +140,8 @@ public <T> T get(final int position, final Class<T> valueClass) {
*/
@Override
public <T> void set(final int position, final T value) {
final RecordField recordField = record.getSchema().getField(position);
record.setValue(recordField, value);
final String fieldName = struct.fields().get(position).name();
record.setValue(fieldName, value);
}

@Override
Expand Down
136 changes: 115 additions & 21 deletions ...g-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,20 +16,33 @@
*/
package org.apache.nifi.processors.iceberg.record;

import org.apache.commons.lang3.ArrayUtils;
import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.variants.Variant;
import org.apache.iceberg.variants.VariantPrimitive;
import org.apache.iceberg.variants.Variants;
import org.apache.nifi.serialization.record.DataType;
import org.apache.nifi.serialization.record.MapRecord;
import org.apache.nifi.serialization.record.Record;
import org.apache.nifi.serialization.record.RecordField;
import org.apache.nifi.serialization.record.RecordFieldType;
import org.apache.nifi.serialization.record.RecordSchema;

import java.math.BigDecimal;
import java.nio.ByteBuffer;
import java.sql.Timestamp;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.sql.Date;
import java.sql.Time;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;

/**
* Record Converter handles translating field values to types compatible with Apache Iceberg Records
Expand All @@ -39,7 +52,11 @@ class RecordConverter {
private static final Set<RecordFieldType> CONVERSION_REQUIRED_FIELD_TYPES = Set.of(
RecordFieldType.TIMESTAMP,
RecordFieldType.DATE,
RecordFieldType.TIME
RecordFieldType.TIME,
RecordFieldType.UUID,
RecordFieldType.ARRAY,
RecordFieldType.RECORD,
RecordFieldType.MAP
);

/**
Expand All @@ -48,44 +65,116 @@ class RecordConverter {
* @param inputRecord Input Record to be converted
* @return Input Record or new Record with converted field values
*/
static Record getConvertedRecord(final Record inputRecord) {
static Record getConvertedRecord(final Record inputRecord, final Types.StructType struct) {
final Record convertedRecord;

final RecordSchema recordSchema = inputRecord.getSchema();
if (isConversionRequired(recordSchema)) {
final Map<String, Object> values = inputRecord.toMap();
convertedRecord = getConvertedRecord(recordSchema, values);
if (isConversionRequired(inputRecord.getSchema(), struct)) {
convertedRecord = convertRecord(inputRecord, struct);
} else {
convertedRecord = inputRecord;
}

return convertedRecord;
}

private static Record getConvertedRecord(final RecordSchema recordSchema, final Map<String, Object> values) {
static Record convertRecord(final Record inputRecord, final Types.StructType struct) {
final RecordSchema recordSchema = inputRecord.getSchema();
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);
for (final RecordField field : recordSchema.getFields()) {
final String fieldName = field.getFieldName();
final Types.NestedField icebergField = struct.field(fieldName);
final Object rawValue = values.get(fieldName);
if (icebergField == null) {
convertedValues.put(fieldName, rawValue);
}
else {
convertedValues.put(fieldName, convertValue(rawValue, icebergField.type()));
}
}

return new MapRecord(recordSchema, convertedValues);
}

private static Object getConvertedValue(final Object value) {
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;
private static Object convertValue(final Object value, final org.apache.iceberg.types.Type icebergType) {
if (value == null) {
return null;
}

return switch (icebergType.typeId()) {
case TIMESTAMP -> {
final LocalDateTime local = ((Timestamp) value).toLocalDateTime();
yield ((Types.TimestampType) icebergType).shouldAdjustToUTC()
? local.atOffset(ZoneOffset.UTC)
: local;
}
case UUID -> value instanceof java.util.UUID ? value : java.util.UUID.fromString(value.toString());
case FIXED -> toByteArray(value);
case BINARY -> ByteBuffer.wrap(toByteArray(value));
case LIST -> {
final Iterable<?> elements = value instanceof Object[] array ? Arrays.asList(array) : (List<?>) value;
final List<Object> list = new ArrayList<>();
for (final Object element : elements) {
list.add(convertValue(element, ((Types.ListType) icebergType).elementType()));
}
yield list;
}
case STRUCT -> value instanceof final org.apache.iceberg.data.Record convertedStruct
? convertedStruct
: new DelegatedRecord(
(org.apache.nifi.serialization.record.Record) value,
(Types.StructType) icebergType
);
case MAP -> {
final Types.MapType mapType = (Types.MapType) icebergType;
final Map<Object, Object> converted = new LinkedHashMap<>();
for (final Map.Entry<Object, Object> entry : ((Map<Object, Object>) value).entrySet()) {
final Object convertedKey = convertValue(entry.getKey(), mapType.keyType());
final Object convertedValue = convertValue(entry.getValue(), mapType.valueType());
converted.put(convertedKey, convertedValue);
}
yield converted;
}
case DATE -> ((Date) value).toLocalDate();
case TIME -> ((Time) value).toLocalTime();
case VARIANT -> convertVariant(value);
default -> value;
};
}

private static boolean isConversionRequired(final RecordSchema recordSchema) {
private static Variant convertVariant(final Object value) {
if (value instanceof Variant variant) {
return variant;
}

final VariantPrimitive<?> primitive = switch (value) {
case Boolean booleanValue -> Variants.of(booleanValue);
case Integer integerValue -> Variants.of(integerValue);
case Long longValue -> Variants.of(longValue);
case Float floatValue -> Variants.of(floatValue);
case Double doubleValue -> Variants.of(doubleValue);
case BigDecimal decimalValue -> Variants.of(decimalValue);
case ByteBuffer byteBufferValue -> Variants.of(byteBufferValue);
case byte[] bytesValue -> Variants.of(ByteBuffer.wrap(bytesValue));
case UUID uuidValue -> Variants.ofUUID(uuidValue);
case String stringValue -> Variants.of(stringValue);
default -> Variants.of(value.toString());
};

return Variant.of(Variants.emptyMetadata(), primitive);
}

private static byte[] toByteArray(final Object value) {
if (value instanceof byte[] bytes) {
return bytes;
}
final Object[] boxed = (Object[]) value;
final Byte[] boxedBytes = Arrays.copyOf(boxed, boxed.length, Byte[].class);
return ArrayUtils.toPrimitive(boxedBytes);
}

private static boolean isConversionRequired(final RecordSchema recordSchema, final Types.StructType struct) {
final List<RecordField> fields = recordSchema.getFields();

for (final RecordField field : fields) {
Expand All @@ -94,6 +183,11 @@ private static boolean isConversionRequired(final RecordSchema recordSchema) {
if (CONVERSION_REQUIRED_FIELD_TYPES.contains(recordFieldType)) {
return true;
}

final Types.NestedField icebergField = struct.field(field.getFieldName());
if (icebergField != null && icebergField.type().typeId() == Type.TypeID.VARIANT) {
return true;
}
}

return false;
Expand Down
51 changes: 49 additions & 2 deletions ...ocessors/src/test/java/org/apache/nifi/processors/iceberg/record/DelegatedRecordTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,7 +82,9 @@ void testSetGetStringField() {

final Record record = new MapRecord(recordSchema, values);

final Types.StructType structType = Types.StructType.of();
final Types.StructType structType = Types.StructType.of(
Types.NestedField.required(1, LABEL_FIELD, Types.StringType.get())
);
final DelegatedRecord delegatedRecord = new DelegatedRecord(record, structType);

final Types.StructType recordStruct = delegatedRecord.struct();
Expand Down Expand Up @@ -123,7 +125,11 @@ void testGetTimestampDateTimeFields() {

final Record record = new MapRecord(recordSchema, values);

final Types.StructType structType = Types.StructType.of();
final Types.StructType structType = Types.StructType.of(
Types.NestedField.required(1, CREATED_FIELD, Types.TimestampType.withoutZone()),
Types.NestedField.required(2, UPDATED_FIELD, Types.DateType.get()),
Types.NestedField.required(3, STOPPED_FIELD, Types.TimeType.get())
);
final DelegatedRecord delegatedRecord = new DelegatedRecord(record, structType);

final Object created = delegatedRecord.getField(CREATED_FIELD);
Expand All @@ -135,4 +141,45 @@ void testGetTimestampDateTimeFields() {
final Object stopped = delegatedRecord.getField(STOPPED_FIELD);
assertEquals(STOPPED_CONVERTED, stopped);
}

@Test
void testCopyRecordWithNestedStruct() {
final String nestedField = "nested";
final String innerTimestampField = "innerTimestamp";
final Timestamp innerTimestamp = Timestamp.valueOf("2026-03-04 08:15:30");

final RecordSchema nestedSchema = new SimpleRecordSchema(
List.of(new RecordField(innerTimestampField, RecordFieldType.TIMESTAMP.getDataType()))
);
final Map<String, Object> nestedValues = new LinkedHashMap<>();
nestedValues.put(innerTimestampField, innerTimestamp);
final Record nestedRecord = new MapRecord(nestedSchema, nestedValues);

final RecordSchema outerSchema = new SimpleRecordSchema(
List.of(
new RecordField(LABEL_FIELD, RecordFieldType.STRING.getDataType()),
new RecordField(nestedField, RecordFieldType.RECORD.getDataType())
)
);
final Map<String, Object> outerValues = new LinkedHashMap<>();
outerValues.put(LABEL_FIELD, LABEL);
outerValues.put(nestedField, nestedRecord);
final Record outerRecord = new MapRecord(outerSchema, outerValues);

final Types.StructType nestedStructType = Types.StructType.of(
Types.NestedField.required(2, innerTimestampField, Types.TimestampType.withoutZone())
);
final Types.StructType outerStructType = Types.StructType.of(
Types.NestedField.required(1, LABEL_FIELD, Types.StringType.get()),
Types.NestedField.required(3, nestedField, nestedStructType)
);

final DelegatedRecord delegatedRecord = new DelegatedRecord(outerRecord, outerStructType);

final org.apache.iceberg.data.Record firstCopy = delegatedRecord.copy();
assertEquals(delegatedRecord, firstCopy);

final org.apache.iceberg.data.Record secondCopy = ((DelegatedRecord) firstCopy).copy();
assertEquals(delegatedRecord, secondCopy);
}
}
Loading