[lake/iceberg] Support RowType conversion in FlussArrayAsIcebergList

[beryllw]

Purpose

Linked issue: close #xxx
Support RowType conversion in FlussArrayAsIcebergList

Iceberg tiering fails when writing tables with ARRAY<ROW<...>> type, throwing UnsupportedOperationException: Unsupported array element type conversion for Fluss type: RowType.

Reproduction：

CREATE TABLE nested_complex_table (
    `id` INT NOT NULL,
    -- ARRAY of ROW
    `contacts` ARRAY<ROW<`type` STRING, `value` STRING>>,
    -- MAP with ARRAY value
    `tags_by_group` MAP<STRING, ARRAY<STRING>>,
    PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
    'table.datalake.enabled' = 'true',
    'table.datalake.freshness' = '30s',
    'table.datalake.format' = 'iceberg'
);
INSERT INTO nested_complex_table VALUES (
    1,
    ARRAY[ROW('email', 'test@example.com'), ROW('phone', '13800000001')],
    MAP['skills', ARRAY['java', 'sql'], 'hobbies', ARRAY['reading']]
);

Exception Stack Trace:

org.apache.flink.runtime.JobException: Recovery is suppressed by NoRestartBackoffTimeStrategy
	at org.apache.flink.runtime.executiongraph.failover.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:219)
	...
Caused by: java.lang.UnsupportedOperationException: Unsupported array element type conversion for Fluss type: RowType
	at org.apache.fluss.lake.iceberg.source.FlussArrayAsIcebergList.get(FlussArrayAsIcebergList.java:119)
	at java.base/java.util.AbstractList$Itr.next(Unknown Source)
	at org.apache.iceberg.parquet.ParquetValueWriters$RepeatedWriter.write(ParquetValueWriters.java:504)
	at org.apache.iceberg.parquet.ParquetValueWriters$OptionWriter.write(ParquetValueWriters.java:421)
	at org.apache.iceberg.parquet.ParquetValueWriters$StructWriter.write(ParquetValueWriters.java:665)
	at org.apache.iceberg.parquet.ParquetWriter.add(ParquetWriter.java:138)
	at org.apache.iceberg.io.DataWriter.write(DataWriter.java:71)
	at org.apache.iceberg.io.BaseTaskWriter$RollingFileWriter.write(BaseTaskWriter.java:401)
	at org.apache.iceberg.io.BaseTaskWriter$RollingFileWriter.write(BaseTaskWriter.java:384)
	at org.apache.iceberg.io.BaseTaskWriter$BaseRollingWriter.write(BaseTaskWriter.java:311)
	at org.apache.iceberg.io.BaseTaskWriter$BaseEqualityDeltaWriter.write(BaseTaskWriter.java:165)
	at org.apache.fluss.lake.iceberg.tiering.writer.GenericRecordDeltaWriter.write(GenericRecordDeltaWriter.java:61)
	at org.apache.fluss.lake.iceberg.tiering.writer.DeltaTaskWriter.write(DeltaTaskWriter.java:49)
	at org.apache.fluss.lake.iceberg.tiering.IcebergLakeWriter.write(IcebergLakeWriter.java:104)
	... 10 more

FlussArrayAsIcebergList.get() was missing the RowType conversion branch. While FlussMapAsIcebergMap.convertElement() already supports RowType conversion (lines 162-169), the array converter lacked this support.

Brief change log

Tests

API and Format

Documentation