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..eff7652be1ae 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; @@ -70,15 +69,17 @@ public void setField(final String fieldName, final Object fieldValue) { } /** - * Get Field value for specified position from supporting Record + * Get Field value for specified position from supporting Record. The position refers to the Iceberg Table struct, + * so the field is resolved by the Iceberg column name to align incoming Record fields with Table columns + * regardless of the incoming Record field ordering. Columns not present in the incoming Record return null. * - * @param position Field position + * @param position Field position in the Iceberg Table struct * @return Field value or null when not found */ @Override public Object get(final int position) { - final RecordField recordField = record.getSchema().getField(position); - return record.getValue(recordField); + final Types.NestedField field = struct.fields().get(position); + return record.getValue(field.name()); } /** @@ -133,16 +134,18 @@ public T get(final int position, final Class valueClass) { } /** - * Set Field value for specified position + * Set Field value for specified position. The position refers to the Iceberg Table struct, so the field is resolved + * by the Iceberg column name to remain symmetric with {@link #get(int)} regardless of the incoming Record field + * ordering. * - * @param position Field position + * @param position Field position in the Iceberg Table struct * @param value Field value * @param Field Value Type */ @Override public void set(final int position, final T value) { - final RecordField recordField = record.getSchema().getField(position); - record.setValue(recordField, value); + final Types.NestedField field = struct.fields().get(position); + record.setValue(field.name(), 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 7e43a5c44489..3b6785213878 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 @@ -25,6 +25,7 @@ import org.apache.nifi.serialization.record.RecordSchema; import org.junit.jupiter.api.Test; +import java.math.BigDecimal; import java.sql.Date; import java.sql.Time; import java.sql.Timestamp; @@ -36,6 +37,7 @@ import java.util.Map; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; class DelegatedRecordTest { @@ -53,6 +55,18 @@ class DelegatedRecordTest { private static final Time STOPPED = Time.valueOf("23:30:45"); private static final LocalTime STOPPED_CONVERTED = STOPPED.toLocalTime(); + private static final String ID_FIELD = "id"; + private static final String AMOUNT_FIELD = "amount"; + private static final int ID = 7; + private static final int ID_MODIFIED = 99; + private static final BigDecimal AMOUNT = new BigDecimal("12.34"); + + private static final Types.StructType TABLE_STRUCT = Types.StructType.of( + Types.NestedField.required(1, ID_FIELD, Types.IntegerType.get()), + Types.NestedField.optional(2, AMOUNT_FIELD, Types.DecimalType.of(10, 2)), + Types.NestedField.optional(3, LABEL_FIELD, Types.StringType.get()) + ); + @Test void testCopyEmptyRecord() { final List recordFields = List.of(); @@ -82,7 +96,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.optional(1, LABEL_FIELD, Types.StringType.get()) + ); final DelegatedRecord delegatedRecord = new DelegatedRecord(record, structType); final Types.StructType recordStruct = delegatedRecord.struct(); @@ -135,4 +151,69 @@ void testGetTimestampDateTimeFields() { final Object stopped = delegatedRecord.getField(STOPPED_FIELD); assertEquals(STOPPED_CONVERTED, stopped); } + + /** + * Iceberg writers read values positionally against the table struct, so position 0 must always return the value of + * the table's first column ("id"), independent of the field ordering in the incoming Record schema. + */ + @Test + void testGetByPositionMatchesTableColumnNameRegardlessOfInputOrder() { + final DelegatedRecord delegatedRecord = new DelegatedRecord(shuffledInputRecord(), TABLE_STRUCT); + + assertEquals(ID, delegatedRecord.get(0)); + assertEquals(AMOUNT, delegatedRecord.get(1)); + assertEquals(LABEL, delegatedRecord.get(2)); + } + + /** + * When the incoming Record does not contain a column present in the table schema, positional access must return + * null for that column rather than shifting subsequent input values into it. + */ + @Test + void testGetByPositionReturnsNullForColumnMissingFromInput() { + final RecordSchema recordSchema = new SimpleRecordSchema(List.of( + new RecordField(ID_FIELD, RecordFieldType.INT.getDataType()), + new RecordField(LABEL_FIELD, RecordFieldType.STRING.getDataType()) + )); + final Map values = new LinkedHashMap<>(); + values.put(ID_FIELD, ID); + values.put(LABEL_FIELD, LABEL); + final DelegatedRecord delegatedRecord = new DelegatedRecord(new MapRecord(recordSchema, values), TABLE_STRUCT); + + assertEquals(ID, delegatedRecord.get(0)); + assertNull(delegatedRecord.get(1), "Missing '" + AMOUNT_FIELD + "' column must be null, not shifted input data"); + assertEquals(LABEL, delegatedRecord.get(2)); + } + + /** + * Positional set must resolve the target field by the Iceberg table column name for the given position, independent + * of the incoming Record field ordering, so that set(position) is symmetric with get(position). + */ + @Test + void testSetByPositionMatchesTableColumnNameRegardlessOfInputOrder() { + final DelegatedRecord delegatedRecord = new DelegatedRecord(shuffledInputRecord(), TABLE_STRUCT); + + delegatedRecord.set(0, ID_MODIFIED); + + assertEquals(ID_MODIFIED, delegatedRecord.getField(ID_FIELD)); + assertEquals(AMOUNT, delegatedRecord.getField(AMOUNT_FIELD)); + assertEquals(LABEL, delegatedRecord.getField(LABEL_FIELD)); + } + + /** + * Build an incoming Record whose field ordering (amount, label, id) intentionally differs from the + * {@link #TABLE_STRUCT} ordering (id, amount, label) so positional access must resolve by column name. + */ + private static Record shuffledInputRecord() { + final RecordSchema recordSchema = new SimpleRecordSchema(List.of( + new RecordField(AMOUNT_FIELD, RecordFieldType.DECIMAL.getDataType()), + new RecordField(LABEL_FIELD, RecordFieldType.STRING.getDataType()), + new RecordField(ID_FIELD, RecordFieldType.INT.getDataType()) + )); + final Map values = new LinkedHashMap<>(); + values.put(AMOUNT_FIELD, AMOUNT); + values.put(LABEL_FIELD, LABEL); + values.put(ID_FIELD, ID); + return new MapRecord(recordSchema, values); + } }