Preserve null lists when projecting a list of structs

pyiceberg/io/pyarrow.py

@@ -2111,7 +2111,14 @@ def list(self, list_type: ListType, list_array: pa.Array | None, value_array: pa
             if isinstance(value_array, pa.StructArray):
                 # This can be removed once this has been fixed:
                 # https://github.com/apache/arrow/issues/38809
-                list_array = pa.LargeListArray.from_arrays(list_array.offsets, value_array)
+                # Pass the validity mask, otherwise null lists become empty lists. Arrow refuses a mask
+                # with sliced offsets, so rebase the offsets onto the values the list covers.
+                start, end = list_array.offsets[0].as_py(), list_array.offsets[-1].as_py()
+                list_array = pa.LargeListArray.from_arrays(
+                    pc.subtract(list_array.offsets, start),
+                    value_array.slice(start, end - start),
+                    mask=list_array.is_null(),
+                )
             value_array = self._cast_if_needed(list_type.element_field, value_array)
             arrow_field = list_initializer(self._construct_field(list_type.element_field, value_array.type))
             return list_array.cast(arrow_field)

tests/integration/test_reads.py

@@ -1001,10 +1001,7 @@ def test_null_list_and_map(catalog: Catalog) -> None:
     arrow_table = table_test_empty_list_and_map.scan().to_arrow()
     assert arrow_table["col_list"].to_pylist() == [None, []]
     assert arrow_table["col_map"].to_pylist() == [None, []]
-    # This should be:
-    # assert arrow_table["col_list_with_struct"].to_pylist() == [None, [{'test': 1}]]
-    # Once https://github.com/apache/arrow/issues/38809 has been fixed
-    assert arrow_table["col_list_with_struct"].to_pylist() == [[], [{"test": 1}]]
+    assert arrow_table["col_list_with_struct"].to_pylist() == [None, [{"test": 1}]]
 
 
 @pytest.mark.integration

tests/io/test_pyarrow.py

@@ -24,7 +24,7 @@
 import uuid
 import warnings
 import zlib
-from collections.abc import Iterator
+from collections.abc import Callable, Iterator
 from datetime import date, datetime, timezone
 from pathlib import Path
 from typing import Any, Literal
@@ -3579,6 +3579,37 @@ def test__to_requested_schema_null_map_of_structs() -> None:
     assert result.column(1).to_pylist() == [[], None, [("b", 2), ("c", None)]]
 
 
+@pytest.mark.parametrize("list_type", [pa.list_, pa.large_list])
+def test__to_requested_schema_null_list_of_structs(list_type: Callable[[pa.DataType], pa.DataType]) -> None:
+    requested_schema = Schema(
+        NestedField(
+            1,
+            "col_list_with_struct",
+            ListType(
+                11,
+                StructType(NestedField(111, "source", StringType(), required=False)),
+                element_required=False,
+            ),
+            required=False,
+        ),
+    )
+    arrow_schema = pa.schema([pa.field("col_list_with_struct", list_type(pa.struct([pa.field("source", pa.string())])))])
+    batch = pa.RecordBatch.from_arrays(
+        [pa.array([[{"source": "x"}], [], None, [{"source": "y"}, None]], type=arrow_schema.field(0).type)],
+        schema=arrow_schema,
+    )
+
+    result = _to_requested_schema(requested_schema, requested_schema, batch)
+    assert result.column(0).to_pylist() == [[{"source": "x"}], [], None, [{"source": "y"}, None]]
+
+    # A sliced batch has to keep the nulls as well
+    result = _to_requested_schema(requested_schema, requested_schema, batch.slice(1, 3))
+    assert result.column(0).to_pylist() == [[], None, [{"source": "y"}, None]]
+
+    result = _to_requested_schema(requested_schema, requested_schema, batch.slice(2, 0))
+    assert result.column(0).to_pylist() == []
+
+
 @pytest.mark.parametrize(
     "arrow_type,iceberg_type,expected_arrow_type",
     [