NIFI-15568 Fix Timestamp Partitioning in PutIcebergRecord

[exceptionfactory]

Summary

NIFI-15568 Corrects partition by Timestamp, Date, and Time fields in the PutIcebergRecord Processor and the ParquetIcebergWriter Controller Service.

Based on the initial approach in #10877, changes include selected conversion of fields with java.sql types to corresponding java.time types in the DelegatedRecord class. The conversion process evaluates the record schema to determine the presence of field types requiring conversion, avoiding unnecessary object creation.

Instead of adding the iceberg-data library, which brings in additional transitive dependencies, changes to the ParquetIcebergWriter include a PartitionKeyRecord for wrapping input Iceberg Record objects and returning primitive partition keys following the pattern of the Iceberg InternalRecordWrapper.

New unit tests verify the behavior of the DelegatedRecord with Timestamp, Date, and Time fields, and also verify partition key handling for ParquetIcebergWriter.

Tracking

Please complete the following tracking steps prior to pull request creation.

Issue Tracking

• Apache NiFi Jira issue created

Pull Request Tracking

• Pull Request title starts with Apache NiFi Jira issue number, such as NIFI-00000
• Pull Request commit message starts with Apache NiFi Jira issue number, as such NIFI-00000
• Pull request contains commits signed with a registered key indicating Verified status

Pull Request Formatting

• Pull Request based on current revision of the main branch
• Pull Request refers to a feature branch with one commit containing changes

Verification

Please indicate the verification steps performed prior to pull request creation.

Build

• Build completed using ./mvnw clean install -P contrib-check
  • JDK 21
  • JDK 25

Licensing

• New dependencies are compatible with the Apache License 2.0 according to the License Policy
• New dependencies are documented in applicable LICENSE and NOTICE files

Documentation

• Documentation formatting appears as expected in rendered files

[pvillard31]

Copy/paste I guess?

Suggested change

	final Types.TimestampType timestampType = (Types.TimestampType) fieldType;
	if (timestampType.shouldAdjustToUTC()) {
	converter = dateTime -> DateTimeUtil.nanosFromTimestamptz((OffsetDateTime) dateTime);
	} else {
	converter = dateTime -> DateTimeUtil.nanosFromTimestamp((LocalDateTime) dateTime);
	}
	final Types.TimestampType timestampType = (Types.TimestampType) fieldType;
	if (timestampType.shouldAdjustToUTC()) {
	converter = dateTime -> DateTimeUtil.microsFromTimestamptz((OffsetDateTime) dateTime);
	} else {
	converter = dateTime -> DateTimeUtil.microsFromTimestamp((LocalDateTime) dateTime);
	}

The existing test testWriteDataFilesPartitionedTimestamp does not catch this because it only verifies that dataFiles has length 1 and a recordCount() of 1, it never asserts on the actual partition values of the output DataFile.

[exceptionfactory]

Good catch, I will adjust and add some testing for this behavior.

[pvillard31]

Do we want to keep those?

[exceptionfactory]

Yes, this is needed for casting the array of functional converters.

[pvillard31]

We do not support nested records. The PartitionKeyRecord does handle nested structs via its STRUCT converter, but not here. Might be an acceptable limitation for know but wanted to point it out.

[exceptionfactory]

Thanks for noting this detail, yes, the PartitionKeyRecord follows the pattern of the implementation from Apache Iceberg, but this RecordConverter is more narrowly implemented. The difference in capability is acceptable for now, but I will track it for subsequent improvement with more general support for nested structures.

[exceptionfactory]

Thanks for the review @pvillard31, I pushed a commit to use the micros conversion method for the Timestamp Type, and added a unit test for non-nested field types of PartitionKeyRecord.