This is an automated email from the ASF dual-hosted git repository.
pvillard31 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 5aa1196f085 NIFI-16074 Added binary, fixed, UUID and geospatial value
conversion to PutIcebergRecord (#11749)
5aa1196f085 is described below
commit 5aa1196f0851bdc3687adcb9bc2173030544a252
Author: Henri Saksi <[email protected]>
AuthorDate: Wed Oct 7 15:08:05 2026 +0300
NIFI-16074 Added binary, fixed, UUID and geospatial value conversion to
PutIcebergRecord (#11749)
---
.../iceberg/parquet/ParquetIcebergWriterTest.java | 130 +++++++++++++++
.../processors/iceberg/record/RecordConverter.java | 70 ++++++++-
.../iceberg/record/DelegatedRecordTest.java | 44 ++++++
.../iceberg/record/RecordConverterTest.java | 174 +++++++++++++++++++++
4 files changed, 412 insertions(+), 6 deletions(-)
diff --git
a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-parquet-writer/src/test/java/org/apache/nifi/services/iceberg/parquet/ParquetIcebergWriterTest.java
b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-parquet-writer/src/test/java/org/apache/nifi/services/iceberg/parquet/ParquetIcebergWriterTest.java
index 1a29da01046..d2fa7cb2b84 100644
---
a/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-parquet-writer/src/test/java/org/apache/nifi/services/iceberg/parquet/ParquetIcebergWriterTest.java
+++
b/nifi-extension-bundles/nifi-iceberg-bundle/nifi-iceberg-parquet-writer/src/test/java/org/apache/nifi/services/iceberg/parquet/ParquetIcebergWriterTest.java
@@ -21,15 +21,22 @@ import org.apache.iceberg.FileFormat;
import org.apache.iceberg.PartitionData;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
+import org.apache.iceberg.SnapshotSummary;
import org.apache.iceberg.StructLike;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.data.GenericRecord;
import org.apache.iceberg.encryption.EncryptedOutputFile;
import org.apache.iceberg.encryption.EncryptionManager;
+import org.apache.iceberg.geospatial.GeospatialBound;
+import org.apache.iceberg.inmemory.InMemoryCatalog;
import org.apache.iceberg.inmemory.InMemoryOutputFile;
import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.io.LocationProvider;
import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.types.Conversions;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.util.DateTimeUtil;
import org.apache.nifi.reporting.InitializationException;
@@ -44,11 +51,14 @@ import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.util.List;
import java.util.Map;
+import java.util.UUID;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
@@ -96,6 +106,48 @@ class ParquetIcebergWriterTest {
private static final Map<String, String> ATTRIBUTES_FIELD_VALUE =
Map.of("k", "v");
+ private static final int DATA_FIELD_ID = 1;
+
+ private static final String DATA_FIELD_NAME = "data";
+
+ private static final int DIGEST_FIELD_ID = 2;
+
+ private static final String DIGEST_FIELD_NAME = "digest";
+
+ private static final int IDENTIFIER_FIELD_ID = 3;
+
+ private static final String IDENTIFIER_FIELD_NAME = "identifier";
+
+ private static final int GEOMETRY_FIELD_ID = 1;
+
+ private static final String GEOMETRY_FIELD_NAME = "geometry";
+
+ private static final int GEOGRAPHY_FIELD_ID = 2;
+
+ private static final String GEOGRAPHY_FIELD_NAME = "geography";
+
+ private static final byte[] DATA_FIELD_VALUE = new byte[] {1, 2, 3, 4};
+
+ private static final UUID IDENTIFIER_FIELD_VALUE =
UUID.fromString("0f8fad5b-d9cb-469f-a165-70867728950e");
+
+ private static final double POINT_LONGITUDE = 24.9384;
+
+ private static final double POINT_LATITUDE = 60.1699;
+
+ // Well-Known Binary encoding of a Point with little-endian byte order,
Point geometry type, and X and Y coordinates
+ private static final byte[] POINT_FIELD_VALUE =
ByteBuffer.allocate(21).order(ByteOrder.LITTLE_ENDIAN)
+ .put((byte) 1)
+ .putInt(1)
+ .putDouble(POINT_LONGITUDE)
+ .putDouble(POINT_LATITUDE)
+ .array();
+
+ private static final String CATALOG_NAME = "memory";
+
+ private static final TableIdentifier GEOSPATIAL_TABLE_IDENTIFIER =
TableIdentifier.of(Namespace.of("default"), "geospatial");
+
+ private static final String FORMAT_VERSION_3 = "3";
+
private ParquetIcebergWriter parquetIcebergWriter;
private TestRunner runner;
@@ -251,6 +303,84 @@ class ParquetIcebergWriterTest {
assertDataFilesFound(dataFiles, serialized);
}
+ /**
+ * Iceberg Parquet writers require ByteBuffer values for binary columns,
byte arrays for fixed columns, and UUID values
+ * for uuid columns, and record the written values as column bounds.
+ */
+ @Test
+ void testWriteDataFilesBinaryFixedUuidTypes() throws IOException {
+ runner.enableControllerService(parquetIcebergWriter);
+
+ final Types.FixedType fixedType =
Types.FixedType.ofLength(DATA_FIELD_VALUE.length);
+ final Schema schema = new Schema(
+ Types.NestedField.optional(DATA_FIELD_ID, DATA_FIELD_NAME,
Types.BinaryType.get()),
+ Types.NestedField.optional(DIGEST_FIELD_ID, DIGEST_FIELD_NAME,
fixedType),
+ Types.NestedField.optional(IDENTIFIER_FIELD_ID,
IDENTIFIER_FIELD_NAME, Types.UUIDType.get())
+ );
+ final InMemoryOutputFile outputFile = new InMemoryOutputFile();
+ final PartitionSpec partitionSpec = PartitionSpec.unpartitioned();
+ setTable(schema, partitionSpec, outputFile);
+
when(locationProvider.newDataLocation(anyString())).thenReturn(LOCATION);
+
+ final IcebergRowWriter rowWriter =
parquetIcebergWriter.getRowWriter(table);
+
+ final GenericRecord row = GenericRecord.create(schema);
+ row.setField(DATA_FIELD_NAME, ByteBuffer.wrap(DATA_FIELD_VALUE));
+ row.setField(DIGEST_FIELD_NAME, DATA_FIELD_VALUE);
+ row.setField(IDENTIFIER_FIELD_NAME, IDENTIFIER_FIELD_VALUE);
+ rowWriter.write(row);
+
+ final DataFile[] dataFiles = rowWriter.dataFiles();
+ final byte[] serialized = outputFile.toByteArray();
+ assertDataFilesFound(dataFiles, serialized);
+
+ final Map<Integer, ByteBuffer> lowerBounds =
dataFiles[0].lowerBounds();
+ assertEquals(ByteBuffer.wrap(DATA_FIELD_VALUE),
Conversions.fromByteBuffer(Types.BinaryType.get(),
lowerBounds.get(DATA_FIELD_ID)));
+ assertEquals(ByteBuffer.wrap(DATA_FIELD_VALUE),
Conversions.fromByteBuffer(fixedType, lowerBounds.get(DIGEST_FIELD_ID)));
+ assertEquals(IDENTIFIER_FIELD_VALUE,
Conversions.fromByteBuffer(Types.UUIDType.get(),
lowerBounds.get(IDENTIFIER_FIELD_ID)));
+ }
+
+ /**
+ * Iceberg geometry and geography types require table format version 3,
and Iceberg Parquet writers require Well-Known
+ * Binary values in ByteBuffers for both types, recording the bounding box
of geometry values as column bounds.
+ */
+ @Test
+ void testWriteDataFilesGeospatialTypes() throws IOException {
+ runner.enableControllerService(parquetIcebergWriter);
+
+ final Types.GeometryType geometryType = Types.GeometryType.crs84();
+ final Schema schema = new Schema(
+ Types.NestedField.optional(GEOMETRY_FIELD_ID,
GEOMETRY_FIELD_NAME, geometryType),
+ Types.NestedField.optional(GEOGRAPHY_FIELD_ID,
GEOGRAPHY_FIELD_NAME, Types.GeographyType.crs84())
+ );
+
+ try (InMemoryCatalog catalog = new InMemoryCatalog()) {
+ catalog.initialize(CATALOG_NAME, Map.of());
+ catalog.createNamespace(GEOSPATIAL_TABLE_IDENTIFIER.namespace());
+ final Table geospatialTable =
catalog.buildTable(GEOSPATIAL_TABLE_IDENTIFIER, schema)
+ .withProperty(TableProperties.FORMAT_VERSION,
FORMAT_VERSION_3)
+ .create();
+
+ final IcebergRowWriter rowWriter =
parquetIcebergWriter.getRowWriter(geospatialTable);
+
+ final GenericRecord row = GenericRecord.create(schema);
+ row.setField(GEOMETRY_FIELD_NAME,
ByteBuffer.wrap(POINT_FIELD_VALUE));
+ row.setField(GEOGRAPHY_FIELD_NAME,
ByteBuffer.wrap(POINT_FIELD_VALUE));
+ rowWriter.write(row);
+
+ final DataFile[] dataFiles = rowWriter.dataFiles();
+ assertEquals(1, dataFiles.length);
+
+ final DataFile dataFile = dataFiles[0];
+ final GeospatialBound pointBound =
GeospatialBound.createXY(POINT_LONGITUDE, POINT_LATITUDE);
+ assertEquals(pointBound, Conversions.fromByteBuffer(geometryType,
dataFile.lowerBounds().get(GEOMETRY_FIELD_ID)));
+ assertEquals(pointBound, Conversions.fromByteBuffer(geometryType,
dataFile.upperBounds().get(GEOMETRY_FIELD_ID)));
+
+ geospatialTable.newAppend().appendFile(dataFile).commit();
+ assertEquals("1",
geospatialTable.currentSnapshot().summary().get(SnapshotSummary.ADDED_RECORDS_PROP));
+ }
+ }
+
private void writeRow(final Schema schema, final IcebergRowWriter
rowWriter) throws IOException {
final GenericRecord row = GenericRecord.create(schema);
row.setField(FIRST_FIELD_NAME, FIRST_FIELD_VALUE);
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 27dad53acb4..bfa9c6d4c64 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
@@ -18,13 +18,16 @@ package org.apache.nifi.processors.iceberg.record;
import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.ByteBuffers;
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 org.apache.nifi.serialization.record.util.DataTypeUtils;
+import java.nio.ByteBuffer;
import java.sql.Date;
import java.sql.Time;
import java.sql.Timestamp;
@@ -64,7 +67,7 @@ class RecordConverter {
final Record convertedRecord;
final RecordSchema recordSchema = inputRecord.getSchema();
- if (isConversionRequired(recordSchema)) {
+ if (isConversionRequired(recordSchema) ||
isUuidConversionRequired(struct)) {
final Map<String, Object> values = inputRecord.toMap();
final Map<String, Object> convertedValues = new
LinkedHashMap<>(values.size());
for (final Map.Entry<String, Object> entry : values.entrySet()) {
@@ -86,7 +89,7 @@ class RecordConverter {
case Timestamp timestamp -> convertTimestamp(timestamp,
icebergType);
case Date date -> date.toLocalDate();
case Time time -> time.toLocalTime();
- // Recursively convert complex types against the matching Iceberg
type
+ // Recursively convert complex types and convert binary and UUID
values against the matching Iceberg type
case null, default -> convertComplexValue(value, icebergType);
};
}
@@ -120,12 +123,13 @@ class RecordConverter {
}
/**
- * Recursively convert array, collection, nested record, and map values
against the matching Iceberg type
+ * Recursively convert array, collection, nested record, and map values
against the matching Iceberg type, and convert
+ * values for Iceberg primitive types that require specific Java types
*
* @param value Field value to be converted
* @param icebergType Iceberg Type describing the target field type (may
be null when not resolved)
- * @return Converted value or the input value when the Iceberg Type is
unknown or does not describe a complex
- * type matching the value
+ * @return Converted value or the input value when the Iceberg Type is
unknown or does not require conversion of the
+ * value
*/
private static Object convertComplexValue(final Object value, final Type
icebergType) {
final Object convertedValue;
@@ -139,7 +143,7 @@ class RecordConverter {
} else if (icebergType.isMapType() && value instanceof Map<?, ?> map) {
convertedValue = convertMap(map, icebergType.asMapType());
} else {
- convertedValue = value;
+ convertedValue = convertPrimitiveValue(value, icebergType);
}
return convertedValue;
@@ -181,6 +185,47 @@ class RecordConverter {
return converted;
}
+ /**
+ * Convert byte array and UUID values to the Java types that Apache
Iceberg requires for the matching primitive type.
+ * Record Readers provide bytes as byte arrays or as Object arrays of Byte
elements, and UUIDs as Strings or byte arrays.
+ *
+ * @param value Field value to be converted
+ * @param icebergType Iceberg Type describing the target field type
+ * @return ByteBuffer for binary, geometry, and geography types, byte
array for fixed types, UUID for uuid types, or the
+ * input value when the Iceberg Type does not require conversion of the
value
+ */
+ private static Object convertPrimitiveValue(final Object value, final Type
icebergType) {
+ return switch (icebergType.typeId()) {
+ // Geometry and geography values are encoded as Well-Known Binary
+ case BINARY, GEOMETRY, GEOGRAPHY -> switch (value) {
+ case byte[] bytes -> ByteBuffer.wrap(bytes);
+ case Object[] array -> ByteBuffer.wrap(toByteArray(array));
+ case null, default -> value;
+ };
+ case FIXED -> switch (value) {
+ case Object[] array -> toByteArray(array);
+ case ByteBuffer buffer -> ByteBuffers.toByteArray(buffer);
+ case null, default -> value;
+ };
+ case UUID -> switch (value) {
+ case String string -> DataTypeUtils.toUUID(string);
+ case byte[] bytes -> DataTypeUtils.toUUID(bytes);
+ case Object[] array ->
DataTypeUtils.toUUID(toByteArray(array));
+ case null, default -> value;
+ };
+ default -> value;
+ };
+ }
+
+ private static byte[] toByteArray(final Object[] array) {
+ final byte[] bytes = new byte[array.length];
+ for (int index = 0; index < array.length; index++) {
+ bytes[index] = (Byte) array[index];
+ }
+
+ return bytes;
+ }
+
private static Type fieldType(final Types.StructType struct, final String
fieldName) {
final Types.NestedField nestedField = struct == null ? null :
struct.field(fieldName);
return nestedField == null ? null : nestedField.type();
@@ -192,4 +237,17 @@ class RecordConverter {
.map(DataType::getFieldType)
.anyMatch(CONVERSION_REQUIRED_FIELD_TYPES::contains);
}
+
+ /**
+ * Determine whether the Iceberg Struct Type contains uuid fields, which
require conversion of String values that do
+ * not otherwise require Record conversion
+ *
+ * @param struct Iceberg Struct Type describing the target field types
(may be null for scalar-only conversion)
+ * @return UUID conversion required status
+ */
+ private static boolean isUuidConversionRequired(final Types.StructType
struct) {
+ return struct != null && struct.fields().stream()
+ .map(Types.NestedField::type)
+ .anyMatch(type -> type.typeId() == Type.TypeID.UUID);
+ }
}
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 3b678521387..5c53983635b 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
@@ -18,6 +18,7 @@ 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.DataType;
import org.apache.nifi.serialization.record.MapRecord;
import org.apache.nifi.serialization.record.Record;
import org.apache.nifi.serialization.record.RecordField;
@@ -26,6 +27,7 @@ import org.apache.nifi.serialization.record.RecordSchema;
import org.junit.jupiter.api.Test;
import java.math.BigDecimal;
+import java.nio.ByteBuffer;
import java.sql.Date;
import java.sql.Time;
import java.sql.Timestamp;
@@ -35,7 +37,9 @@ import java.time.LocalTime;
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.assertNull;
@@ -61,6 +65,13 @@ class DelegatedRecordTest {
private static final int ID_MODIFIED = 99;
private static final BigDecimal AMOUNT = new BigDecimal("12.34");
+ private static final String DATA_FIELD = "data";
+ private static final String DIGEST_FIELD = "digest";
+ private static final String IDENTIFIER_FIELD = "identifier";
+ private static final byte[] DATA = new byte[] {1, 2, 3, 4};
+ private static final Object[] DATA_ELEMENTS = new Object[] {(byte) 1,
(byte) 2, (byte) 3, (byte) 4};
+ private static final UUID IDENTIFIER =
UUID.fromString("0f8fad5b-d9cb-469f-a165-70867728950e");
+
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)),
@@ -152,6 +163,39 @@ class DelegatedRecordTest {
assertEquals(STOPPED_CONVERTED, stopped);
}
+ /**
+ * Record Readers provide bytes as Object arrays of Byte elements and
UUIDs as Strings, while Iceberg writers require
+ * ByteBuffer values for binary columns, byte arrays for fixed columns,
and UUID values for uuid columns.
+ */
+ @Test
+ void testGetBinaryFixedUuidFields() {
+ final DataType bytesDataType =
RecordFieldType.ARRAY.getArrayDataType(RecordFieldType.BYTE.getDataType());
+ final RecordSchema recordSchema = new SimpleRecordSchema(
+ List.of(
+ new RecordField(DATA_FIELD, bytesDataType),
+ new RecordField(DIGEST_FIELD, bytesDataType),
+ new RecordField(IDENTIFIER_FIELD,
RecordFieldType.STRING.getDataType())
+ )
+ );
+ final Map<String, Object> values = new LinkedHashMap<>();
+ values.put(DATA_FIELD, DATA_ELEMENTS);
+ values.put(DIGEST_FIELD, DATA_ELEMENTS);
+ values.put(IDENTIFIER_FIELD, IDENTIFIER.toString());
+
+ final Record record = new MapRecord(recordSchema, values);
+
+ final Types.StructType structType = Types.StructType.of(
+ Types.NestedField.optional(1, DATA_FIELD,
Types.BinaryType.get()),
+ Types.NestedField.optional(2, DIGEST_FIELD,
Types.FixedType.ofLength(DATA.length)),
+ Types.NestedField.optional(3, IDENTIFIER_FIELD,
Types.UUIDType.get())
+ );
+ final DelegatedRecord delegatedRecord = new DelegatedRecord(record,
structType);
+
+ assertEquals(ByteBuffer.wrap(DATA), delegatedRecord.get(0,
ByteBuffer.class));
+ assertArrayEquals(DATA, delegatedRecord.get(1, byte[].class));
+ assertEquals(IDENTIFIER, delegatedRecord.get(2, UUID.class));
+ }
+
/**
* 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.
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 3dcc0fcbf2f..d7cd2c0f643 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
@@ -30,6 +30,7 @@ import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
+import java.nio.ByteBuffer;
import java.sql.Timestamp;
import java.time.LocalDate;
import java.time.LocalDateTime;
@@ -38,10 +39,13 @@ import java.time.ZoneOffset;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.UUID;
import java.util.stream.Stream;
+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.assertNull;
class RecordConverterTest {
@@ -57,12 +61,20 @@ class RecordConverterTest {
private static final String ID_FIELD_NAME = "id";
+ private static final String DATA_FIELD_NAME = "data";
+
+ private static final String IDENTIFIER_FIELD_NAME = "identifier";
+
private static final String CITY_FIELD_VALUE = "Berlin";
private static final String NAME_FIELD_VALUE = "widget";
private static final String ID_FIELD_VALUE = "row-1";
+ private static final byte[] DATA_FIELD_VALUE = new byte[] {1, 2, 3, 4};
+
+ private static final UUID IDENTIFIER_FIELD_VALUE =
UUID.fromString("0f8fad5b-d9cb-469f-a165-70867728950e");
+
private static final LocalDateTime CREATED_LOCAL_DATE_TIME =
LocalDateTime.of(2026, 1, 1, 12, 30, 45);
@Test
@@ -298,4 +310,166 @@ class RecordConverterTest {
assertEquals(CITY_FIELD_VALUE, addressStruct.get(0, String.class));
assertEquals(ID_FIELD_VALUE, converted.getField(ID_FIELD_NAME));
}
+
+ /**
+ * Record Readers provide bytes as byte arrays or as Object arrays of Byte
elements, and Iceberg binary, geometry, and
+ * geography columns require ByteBuffer values.
+ */
+ @ParameterizedTest
+ @MethodSource
+ void testConvertBytesToByteBuffer(final Type icebergType, final Object
value) {
+ final Object converted = RecordConverter.convertValue(value,
icebergType);
+
+ final ByteBuffer buffer = assertInstanceOf(ByteBuffer.class,
converted);
+ assertEquals(ByteBuffer.wrap(DATA_FIELD_VALUE), buffer);
+ }
+
+ private static Stream<Arguments> testConvertBytesToByteBuffer() {
+ return Stream.of(Types.BinaryType.get(), Types.GeometryType.crs84(),
Types.GeographyType.crs84())
+ .flatMap(icebergType -> Stream.of(
+ Arguments.of(icebergType, DATA_FIELD_VALUE.clone()),
+ Arguments.of(icebergType,
getObjectArray(DATA_FIELD_VALUE)),
+ Arguments.of(icebergType,
ByteBuffer.wrap(DATA_FIELD_VALUE))
+ ));
+ }
+
+ /**
+ * Iceberg fixed columns require byte arrays.
+ */
+ @ParameterizedTest
+ @MethodSource
+ void testConvertBytesToFixed(final Object value) {
+ final Types.FixedType fixedType =
Types.FixedType.ofLength(DATA_FIELD_VALUE.length);
+
+ final Object converted = RecordConverter.convertValue(value,
fixedType);
+
+ final byte[] bytes = assertInstanceOf(byte[].class, converted);
+ assertArrayEquals(DATA_FIELD_VALUE, bytes);
+ }
+
+ private static Stream<Arguments> testConvertBytesToFixed() {
+ return Stream.of(
+ Arguments.of(DATA_FIELD_VALUE.clone()),
+ Arguments.of((Object) getObjectArray(DATA_FIELD_VALUE)),
+ Arguments.of(ByteBuffer.wrap(DATA_FIELD_VALUE))
+ );
+ }
+
+ /**
+ * Iceberg uuid columns require UUID values, which Record Readers can
provide as Strings or as byte arrays.
+ */
+ @ParameterizedTest
+ @MethodSource
+ void testConvertUuid(final Object value) {
+ final Object converted = RecordConverter.convertValue(value,
Types.UUIDType.get());
+
+ assertEquals(IDENTIFIER_FIELD_VALUE, converted);
+ }
+
+ private static Stream<Arguments> testConvertUuid() {
+ final byte[] identifierBytes = ByteBuffer.allocate(16)
+ .putLong(IDENTIFIER_FIELD_VALUE.getMostSignificantBits())
+ .putLong(IDENTIFIER_FIELD_VALUE.getLeastSignificantBits())
+ .array();
+
+ return Stream.of(
+ Arguments.of(IDENTIFIER_FIELD_VALUE),
+ Arguments.of(IDENTIFIER_FIELD_VALUE.toString()),
+ Arguments.of(identifierBytes),
+ Arguments.of((Object) getObjectArray(identifierBytes))
+ );
+ }
+
+ @ParameterizedTest
+ @MethodSource
+ void testConvertNullBytesUuid(final Type icebergType) {
+ final Object converted = RecordConverter.convertValue(null,
icebergType);
+
+ assertNull(converted);
+ }
+
+ private static Stream<Arguments> testConvertNullBytesUuid() {
+ return Stream.of(
+ Arguments.of(Types.BinaryType.get()),
+
Arguments.of(Types.FixedType.ofLength(DATA_FIELD_VALUE.length)),
+ Arguments.of(Types.UUIDType.get()),
+ Arguments.of(Types.GeometryType.crs84()),
+ Arguments.of(Types.GeographyType.crs84())
+ );
+ }
+
+ @Test
+ void testConvertArrayElementBytes() {
+ final Types.ListType listType = Types.ListType.ofOptional(1,
Types.BinaryType.get());
+ final Object[] array = new Object[] {getObjectArray(DATA_FIELD_VALUE)};
+
+ final Object converted = RecordConverter.convertValue(array, listType);
+
+ final List<?> list = assertInstanceOf(List.class, converted);
+ assertEquals(List.of(ByteBuffer.wrap(DATA_FIELD_VALUE)), list);
+ }
+
+ @Test
+ void testConvertMapBytesValue() {
+ final Types.MapType mapType = Types.MapType.ofOptional(
+ 1, 2, Types.StringType.get(), Types.BinaryType.get()
+ );
+ final Map<String, Object> map = new LinkedHashMap<>();
+ map.put(DATA_FIELD_NAME, getObjectArray(DATA_FIELD_VALUE));
+
+ final Object converted = RecordConverter.convertValue(map, mapType);
+
+ final Map<?, ?> resultMap = assertInstanceOf(Map.class, converted);
+ assertEquals(ByteBuffer.wrap(DATA_FIELD_VALUE),
resultMap.get(DATA_FIELD_NAME));
+ }
+
+ @Test
+ void testConvertNestedRecordBytesField() {
+ final Types.StructType structType = Types.StructType.of(
+ Types.NestedField.optional(1, DATA_FIELD_NAME,
Types.BinaryType.get())
+ );
+
+ final RecordSchema nestedSchema = new SimpleRecordSchema(List.of(
+ new RecordField(DATA_FIELD_NAME,
RecordFieldType.ARRAY.getArrayDataType(RecordFieldType.BYTE.getDataType()))
+ ));
+ final Map<String, Object> nestedValues = new LinkedHashMap<>();
+ nestedValues.put(DATA_FIELD_NAME, getObjectArray(DATA_FIELD_VALUE));
+ final Record nestedRecord = new MapRecord(nestedSchema, nestedValues);
+
+ final Object converted = RecordConverter.convertValue(nestedRecord,
structType);
+
+ final StructLike struct = assertInstanceOf(StructLike.class,
converted);
+ assertEquals(ByteBuffer.wrap(DATA_FIELD_VALUE), struct.get(0,
ByteBuffer.class));
+ }
+
+ /**
+ * UUID values read as Strings must be converted for Iceberg uuid columns,
even when no other Record field requires
+ * conversion.
+ */
+ @Test
+ void testGetConvertedRecordUuidString() {
+ final Types.StructType struct = Types.StructType.of(
+ Types.NestedField.optional(1, IDENTIFIER_FIELD_NAME,
Types.UUIDType.get())
+ );
+
+ final RecordSchema schema = new SimpleRecordSchema(List.of(
+ new RecordField(IDENTIFIER_FIELD_NAME,
RecordFieldType.STRING.getDataType())
+ ));
+ final Map<String, Object> values = new LinkedHashMap<>();
+ values.put(IDENTIFIER_FIELD_NAME, IDENTIFIER_FIELD_VALUE.toString());
+ final Record record = new MapRecord(schema, values);
+
+ final Record converted = RecordConverter.getConvertedRecord(record,
struct);
+
+ assertEquals(IDENTIFIER_FIELD_VALUE,
converted.getValue(IDENTIFIER_FIELD_NAME));
+ }
+
+ private static Object[] getObjectArray(final byte[] bytes) {
+ final Object[] array = new Object[bytes.length];
+ for (int index = 0; index < bytes.length; index++) {
+ array[index] = bytes[index];
+ }
+
+ return array;
+ }
}
