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..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; @@ -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); } @@ -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/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..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 @@ -16,6 +16,12 @@ */ 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; @@ -23,13 +29,20 @@ 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 @@ -39,7 +52,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 +65,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(), struct)) { + convertedRecord = convertRecord(inputRecord, struct); } else { convertedRecord = inputRecord; } @@ -62,30 +77,104 @@ 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(); + 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 fields = recordSchema.getFields(); for (final RecordField field : fields) { @@ -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; 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..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(); @@ -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); @@ -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 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..e9e8045d6461 --- /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,255 @@ +/* + * 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.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; +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.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 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"; + 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); + } +}