From 4021492e1e091b21c71229947fe2a91a327eb3df Mon Sep 17 00:00:00 2001 From: Nir Yanay Date: Mon, 13 Jul 2026 09:53:51 +0300 Subject: [PATCH 1/3] NIFI-16074 Added conversions for more Iceberg types --- .../iceberg/record/DelegatedRecord.java | 2 +- .../iceberg/record/RecordConverter.java | 100 ++++++-- .../iceberg/record/DelegatedRecordTest.java | 47 +++- .../iceberg/record/RecordConverterTest.java | 226 ++++++++++++++++++ 4 files changed, 353 insertions(+), 22 deletions(-) create mode 100644 nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/RecordConverterTest.java diff --git a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/DelegatedRecord.java b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/DelegatedRecord.java index 830377e19f6c..18cfdc959fe6 100644 --- a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/DelegatedRecord.java +++ b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/DelegatedRecord.java @@ -38,7 +38,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); } diff --git a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java index b84159d4af0a..5796340a13fc 100644 --- a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java +++ b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java @@ -16,6 +16,8 @@ */ package org.apache.nifi.processors.iceberg.record; +import org.apache.commons.lang3.ArrayUtils; +import org.apache.iceberg.types.Types; import org.apache.nifi.serialization.record.DataType; import org.apache.nifi.serialization.record.MapRecord; import org.apache.nifi.serialization.record.Record; @@ -23,9 +25,14 @@ import org.apache.nifi.serialization.record.RecordFieldType; import org.apache.nifi.serialization.record.RecordSchema; +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; @@ -39,7 +46,11 @@ class RecordConverter { private static final Set CONVERSION_REQUIRED_FIELD_TYPES = Set.of( RecordFieldType.TIMESTAMP, RecordFieldType.DATE, - RecordFieldType.TIME + RecordFieldType.TIME, + RecordFieldType.UUID, + RecordFieldType.ARRAY, + RecordFieldType.RECORD, + RecordFieldType.MAP ); /** @@ -48,13 +59,11 @@ 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 values = inputRecord.toMap(); - convertedRecord = getConvertedRecord(recordSchema, values); + if (isConversionRequired(inputRecord.getSchema())) { + convertedRecord = convertRecord(inputRecord, struct); } else { convertedRecord = inputRecord; } @@ -62,29 +71,80 @@ static Record getConvertedRecord(final Record inputRecord) { return convertedRecord; } - private static Record getConvertedRecord(final RecordSchema recordSchema, final Map values) { + static Record convertRecord(final Record inputRecord, final Types.StructType struct) { + final RecordSchema recordSchema = inputRecord.getSchema(); + final Map values = inputRecord.toMap(); final Map convertedValues = new LinkedHashMap<>(); - for (final Map.Entry 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 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 converted = new LinkedHashMap<>(); + for (final Map.Entry entry : ((Map) 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(); + default -> value; }; } + 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 List fields = recordSchema.getFields(); diff --git a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/DelegatedRecordTest.java b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/DelegatedRecordTest.java index 7e43a5c44489..432e5548adbf 100644 --- a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/DelegatedRecordTest.java +++ b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/DelegatedRecordTest.java @@ -123,7 +123,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); @@ -135,4 +139,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 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 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); + } } diff --git a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/RecordConverterTest.java b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/RecordConverterTest.java new file mode 100644 index 000000000000..eff606577150 --- /dev/null +++ b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/RecordConverterTest.java @@ -0,0 +1,226 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.processors.iceberg.record; + +import org.apache.iceberg.types.Types; +import org.apache.nifi.serialization.SimpleRecordSchema; +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 org.junit.jupiter.api.Test; + +import java.nio.ByteBuffer; +import java.sql.Date; +import java.sql.Time; +import java.sql.Timestamp; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.LocalTime; +import java.time.OffsetDateTime; +import java.time.ZoneOffset; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertSame; + +class RecordConverterTest { + + private static final String FIELD_NAME = "field"; + + @Test + void testConvertTimestampWithoutZone() { + final Timestamp timestamp = Timestamp.valueOf("2026-01-01 10:00:00"); + final Types.StructType struct = Types.StructType.of( + Types.NestedField.required(1, FIELD_NAME, Types.TimestampType.withoutZone()) + ); + + final Record converted = convertSingleField(RecordFieldType.TIMESTAMP, timestamp, struct); + + assertEquals(timestamp.toLocalDateTime(), converted.getValue(FIELD_NAME)); + } + + @Test + void testConvertTimestampWithZone() { + final Timestamp timestamp = Timestamp.valueOf("2026-01-01 10:00:00"); + final Types.StructType struct = Types.StructType.of( + Types.NestedField.required(1, FIELD_NAME, Types.TimestampType.withZone()) + ); + + final Record converted = convertSingleField(RecordFieldType.TIMESTAMP, timestamp, struct); + + final OffsetDateTime expected = timestamp.toLocalDateTime().atOffset(ZoneOffset.UTC); + assertEquals(expected, converted.getValue(FIELD_NAME)); + } + + @Test + void testConvertDate() { + final Date date = Date.valueOf("2026-02-03"); + final Types.StructType struct = Types.StructType.of( + Types.NestedField.required(1, FIELD_NAME, Types.DateType.get()) + ); + + final Record converted = convertSingleField(RecordFieldType.DATE, date, struct); + + assertEquals(date.toLocalDate(), converted.getValue(FIELD_NAME)); + } + + @Test + void testConvertTime() { + final Time time = Time.valueOf("23:30:45"); + final Types.StructType struct = Types.StructType.of( + Types.NestedField.required(1, FIELD_NAME, Types.TimeType.get()) + ); + + final Record converted = convertSingleField(RecordFieldType.TIME, time, struct); + + assertEquals(time.toLocalTime(), converted.getValue(FIELD_NAME)); + } + + @Test + void testConvertUuidPassesThroughExistingUuid() { + final UUID uuid = UUID.randomUUID(); + final Types.StructType struct = Types.StructType.of( + Types.NestedField.required(1, FIELD_NAME, Types.UUIDType.get()) + ); + + final Record converted = convertSingleField(RecordFieldType.UUID, uuid, struct); + + assertSame(uuid, converted.getValue(FIELD_NAME)); + } + + @Test + void testConvertUuidFromString() { + final UUID uuid = UUID.randomUUID(); + final Types.StructType struct = Types.StructType.of( + Types.NestedField.required(1, FIELD_NAME, Types.UUIDType.get()) + ); + + final Record converted = convertSingleField(RecordFieldType.UUID, uuid.toString(), struct); + + assertEquals(uuid, converted.getValue(FIELD_NAME)); + } + + @Test + void testConvertFixed() { + final Object[] boxedBytes = {(byte) 1, (byte) 2, (byte) 3}; + final Types.StructType struct = Types.StructType.of( + Types.NestedField.required(1, FIELD_NAME, Types.FixedType.ofLength(3)) + ); + + final Record converted = convertSingleField(RecordFieldType.ARRAY, boxedBytes, struct); + + assertArrayEquals(new byte[] {1, 2, 3}, (byte[]) converted.getValue(FIELD_NAME)); + } + + @Test + void testConvertBinary() { + final Object[] boxedBytes = {(byte) 4, (byte) 5}; + final Types.StructType struct = Types.StructType.of( + Types.NestedField.required(1, FIELD_NAME, Types.BinaryType.get()) + ); + + final Record converted = convertSingleField(RecordFieldType.ARRAY, boxedBytes, struct); + + assertEquals(ByteBuffer.wrap(new byte[] {4, 5}), converted.getValue(FIELD_NAME)); + } + + @Test + void testConvertListRecursivelyConvertsElements() { + final Timestamp first = Timestamp.valueOf("2026-01-01 01:00:00"); + final Timestamp second = Timestamp.valueOf("2026-01-02 02:00:00"); + final Object[] values = {first, second}; + final Types.StructType struct = Types.StructType.of( + Types.NestedField.required(1, FIELD_NAME, + Types.ListType.ofRequired(2, Types.TimestampType.withoutZone())) + ); + + final Record converted = convertSingleField(RecordFieldType.ARRAY, values, struct); + + final List list = (List) converted.getValue(FIELD_NAME); + assertEquals(List.of(first.toLocalDateTime(), second.toLocalDateTime()), list); + } + + @Test + void testConvertMapRecursivelyConvertsValues() { + final Timestamp timestamp = Timestamp.valueOf("2026-01-01 01:00:00"); + final Map value = new LinkedHashMap<>(); + value.put("key", timestamp); + final Types.StructType struct = Types.StructType.of( + Types.NestedField.required(1, FIELD_NAME, + Types.MapType.ofRequired(2, 3, Types.StringType.get(), Types.TimestampType.withoutZone())) + ); + + final Record converted = convertSingleField(RecordFieldType.MAP, value, struct); + + final Map convertedMap = (Map) converted.getValue(FIELD_NAME); + assertEquals(timestamp.toLocalDateTime(), convertedMap.get("key")); + } + + @Test + void testConvertStructWrapsNestedRecordAndConvertsItsFields() { + final String innerField = "innerTimestamp"; + final Timestamp innerTimestamp = Timestamp.valueOf("2026-03-04 08:15:30"); + + final RecordSchema nestedSchema = new SimpleRecordSchema( + List.of(new RecordField(innerField, RecordFieldType.TIMESTAMP.getDataType())) + ); + final Map nestedValues = new LinkedHashMap<>(); + nestedValues.put(innerField, innerTimestamp); + final Record nestedRecord = new MapRecord(nestedSchema, nestedValues); + + final Types.StructType nestedStruct = Types.StructType.of( + Types.NestedField.required(2, innerField, Types.TimestampType.withoutZone()) + ); + final Types.StructType struct = Types.StructType.of( + Types.NestedField.required(1, FIELD_NAME, nestedStruct) + ); + + final Record converted = convertSingleField(RecordFieldType.RECORD, nestedRecord, struct); + + final Object structValue = converted.getValue(FIELD_NAME); + final org.apache.iceberg.data.Record delegatedRecord = assertInstanceOf(org.apache.iceberg.data.Record.class, structValue); + assertEquals(innerTimestamp.toLocalDateTime(), delegatedRecord.getField(innerField)); + } + + @Test + void testFieldWithoutMatchingIcebergFieldPassesThroughUnchanged() { + final String value = "unmapped-value"; + final Types.StructType struct = Types.StructType.of(); + + final Record converted = convertSingleField(RecordFieldType.STRING, value, struct); + + assertEquals(value, converted.getValue(FIELD_NAME)); + } + + private static Record convertSingleField(final RecordFieldType fieldType, final Object value, final Types.StructType struct) { + final RecordSchema recordSchema = new SimpleRecordSchema( + List.of(new RecordField(FIELD_NAME, fieldType.getDataType())) + ); + final Map values = new LinkedHashMap<>(); + values.put(FIELD_NAME, value); + final Record record = new MapRecord(recordSchema, values); + + return RecordConverter.getConvertedRecord(record, struct); + } +} From 6e9d30e8676326e1f5b969327bb97df3d1c9a8d3 Mon Sep 17 00:00:00 2001 From: Nir Yanay Date: Mon, 13 Jul 2026 11:03:58 +0300 Subject: [PATCH 2/3] NIFI-16074 Fix DelegatedRecord positional get/set to resolve by Iceberg struct order Position-based access previously indexed the NiFi record's field order instead of the Iceberg struct's field order, causing values to be mismatched and cast to the wrong type when the two orders diverged. --- .../nifi/processors/iceberg/record/DelegatedRecord.java | 9 ++++----- .../processors/iceberg/record/DelegatedRecordTest.java | 4 +++- 2 files changed, 7 insertions(+), 6 deletions(-) diff --git a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/DelegatedRecord.java b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/DelegatedRecord.java index 18cfdc959fe6..d4f367c93437 100644 --- a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/DelegatedRecord.java +++ b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/DelegatedRecord.java @@ -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; @@ -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); } /** @@ -141,8 +140,8 @@ public T get(final int position, final Class valueClass) { */ @Override public 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 diff --git a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/DelegatedRecordTest.java b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/DelegatedRecordTest.java index 432e5548adbf..ef0edf377fea 100644 --- a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/DelegatedRecordTest.java +++ b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/DelegatedRecordTest.java @@ -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(); From a9f2bb91402dd679f254324379863cb948ff16d9 Mon Sep 17 00:00:00 2001 From: Nir Yanay Date: Sun, 19 Jul 2026 14:21:44 +0300 Subject: [PATCH 3/3] Added variant type support Signed-off-by: Nir Yanay --- .../iceberg/record/RecordConverter.java | 38 ++++++++++++++++++- .../iceberg/record/RecordConverterTest.java | 31 ++++++++++++++- 2 files changed, 66 insertions(+), 3 deletions(-) diff --git a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java index 5796340a13fc..91663ec0818f 100644 --- a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java +++ b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/record/RecordConverter.java @@ -17,7 +17,11 @@ 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; @@ -25,6 +29,7 @@ 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; @@ -37,6 +42,7 @@ 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 @@ -62,7 +68,7 @@ class RecordConverter { static Record getConvertedRecord(final Record inputRecord, final Types.StructType struct) { final Record convertedRecord; - if (isConversionRequired(inputRecord.getSchema())) { + if (isConversionRequired(inputRecord.getSchema(), struct)) { convertedRecord = convertRecord(inputRecord, struct); } else { convertedRecord = inputRecord; @@ -132,10 +138,33 @@ private static Object convertValue(final Object value, final org.apache.iceberg. } case DATE -> ((Date) value).toLocalDate(); case TIME -> ((Time) value).toLocalTime(); + case VARIANT -> convertVariant(value); default -> value; }; } + 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; @@ -145,7 +174,7 @@ private static byte[] toByteArray(final Object value) { return ArrayUtils.toPrimitive(boxedBytes); } - private static boolean isConversionRequired(final RecordSchema recordSchema) { + private static boolean isConversionRequired(final RecordSchema recordSchema, final Types.StructType struct) { final List fields = recordSchema.getFields(); for (final RecordField field : fields) { @@ -154,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; diff --git a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/RecordConverterTest.java b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/RecordConverterTest.java index eff606577150..e9e8045d6461 100644 --- a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/RecordConverterTest.java +++ b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/test/java/org/apache/nifi/processors/iceberg/record/RecordConverterTest.java @@ -17,6 +17,8 @@ package org.apache.nifi.processors.iceberg.record; import org.apache.iceberg.types.Types; +import org.apache.iceberg.variants.Variant; +import org.apache.iceberg.variants.PhysicalType; import org.apache.nifi.serialization.SimpleRecordSchema; import org.apache.nifi.serialization.record.MapRecord; import org.apache.nifi.serialization.record.Record; @@ -31,7 +33,6 @@ import java.sql.Timestamp; import java.time.LocalDate; import java.time.LocalDateTime; -import java.time.LocalTime; import java.time.OffsetDateTime; import java.time.ZoneOffset; import java.util.LinkedHashMap; @@ -203,6 +204,34 @@ void testConvertStructWrapsNestedRecordAndConvertsItsFields() { assertEquals(innerTimestamp.toLocalDateTime(), delegatedRecord.getField(innerField)); } + @Test + void testConvertVariantWrapsPlainStringEvenWhenSourceTypeNeedsNoOtherConversion() { + final Types.StructType struct = Types.StructType.of( + Types.NestedField.required(1, FIELD_NAME, Types.VariantType.get()) + ); + + // RecordFieldType.STRING is not in CONVERSION_REQUIRED_FIELD_TYPES; the VARIANT + // destination type alone must trigger conversion so the writer never sees a raw String. + final Record converted = convertSingleField(RecordFieldType.STRING, "plain-value", struct); + + final Variant variant = assertInstanceOf(Variant.class, converted.getValue(FIELD_NAME)); + assertEquals(PhysicalType.STRING, variant.value().type()); + } + + @Test + void testConvertVariantPassesThroughExistingVariant() { + final Types.StructType struct = Types.StructType.of( + Types.NestedField.required(1, FIELD_NAME, Types.VariantType.get()) + ); + final Variant existing = Variant.of( + org.apache.iceberg.variants.Variants.emptyMetadata(), + org.apache.iceberg.variants.Variants.of(42)); + + final Record converted = convertSingleField(RecordFieldType.STRING, existing, struct); + + assertSame(existing, converted.getValue(FIELD_NAME)); + } + @Test void testFieldWithoutMatchingIcebergFieldPassesThroughUnchanged() { final String value = "unmapped-value";