This is an automated email from the ASF dual-hosted git repository.
exceptionfactory pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new e0242201b20 NIFI-16067 Adjusted PutIcebergRecord to map fields by
ordinal position instead of column name (#11387)
e0242201b20 is described below
commit e0242201b20603f89fe4505a858633fe363a427c
Author: maltesander <[email protected]>
AuthorDate: Sat Jul 25 20:05:25 2026 +0200
NIFI-16067 Adjusted PutIcebergRecord to map fields by ordinal position
instead of column name (#11387)
Signed-off-by: David Handermann <[email protected]>
---
.../processors/iceberg/record/DelegatedRecord.java | 21 +++---
.../iceberg/record/DelegatedRecordTest.java | 83 +++++++++++++++++++++-
2 files changed, 94 insertions(+), 10 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 830377e19f6..eff7652be1a 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 @@ package org.apache.nifi.processors.iceberg.record;
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 class DelegatedRecord implements Record {
}
/**
- * 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 class DelegatedRecord implements Record {
}
/**
- * 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 <T> Field Value Type
*/
@Override
public <T> 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 7e43a5c4448..3b678521387 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.RecordFieldType;
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.List;
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<RecordField> recordFields = List.of();
@@ -82,7 +96,9 @@ class DelegatedRecordTest {
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 @@ class DelegatedRecordTest {
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<String, Object> 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<String, Object> values = new LinkedHashMap<>();
+ values.put(AMOUNT_FIELD, AMOUNT);
+ values.put(LABEL_FIELD, LABEL);
+ values.put(ID_FIELD, ID);
+ return new MapRecord(recordSchema, values);
+ }
}
