[IcebergIO] Fix Iceberg list conversion (#34955)

* array instead of iterable

* cleanup; trigger tests

* fix tests

Author: ahmedabu98

.github/trigger_files/IO_Iceberg_Integration_Tests.json

@@ -1,4 +1,4 @@
 {
     "comment": "Modify this file in a trivial way to cause this test suite to run.",
-    "modification": 2
+    "modification": 3
 }

.github/trigger_files/beam_PostCommit_Python_Xlang_IO_Direct.json

@@ -1,4 +1,4 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "modification": 3
+  "modification": 4
 }

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergUtils.java

@@ -101,8 +101,7 @@ private static Schema.FieldType icebergTypeToBeamFieldType(final Type type) {
       case STRUCT:
         return Schema.FieldType.row(icebergStructTypeToBeamSchema(type.asStructType()));
       case LIST:
-        return Schema.FieldType.iterable(
-            icebergTypeToBeamFieldType(type.asListType().elementType()));
+        return Schema.FieldType.array(icebergTypeToBeamFieldType(type.asListType().elementType()));
       case MAP:
         return Schema.FieldType.map(
             icebergTypeToBeamFieldType(type.asMapType().keyType()),

sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergUtilsTest.java

@@ -700,9 +700,9 @@ public void testPrimitiveIcebergSchemaToBeamSchema() {
 
     static final Schema BEAM_SCHEMA_LIST =
         Schema.builder()
-            .addIterableField("arr_str", Schema.FieldType.STRING)
-            .addIterableField("arr_int", Schema.FieldType.INT32)
-            .addIterableField("arr_bool", Schema.FieldType.BOOLEAN)
+            .addArrayField("arr_str", Schema.FieldType.STRING)
+            .addArrayField("arr_int", Schema.FieldType.INT32)
+            .addArrayField("arr_bool", Schema.FieldType.BOOLEAN)
             .build();
     static final org.apache.iceberg.Schema ICEBERG_SCHEMA_LIST =
         new org.apache.iceberg.Schema(

sdks/python/apache_beam/transforms/managed_iceberg_it_test.py

@@ -48,7 +48,8 @@ def _create_row(self, num: int):
         str_=str(num),
         bytes_=bytes(num),
         bool_=(num % 2 == 0),
-        float_=(num + float(num) / 100))
+        float_=(num + float(num) / 100),
+        arr_=[num, num, num])
 
   def test_write_read_pipeline(self):
     iceberg_config = {