PyArrow: Don't enforce the schema

[Fokko]

PyIceberg struggled with different types of arrow, such as the string and large_string. They represent the same but are different under the hood.

My take is that we should hide this kind of detail from the user as much as possible. Now we went down the road of passing in the Iceberg schema into Arrow, but when doing this, Iceberg has to decide if it is a large or non-large type.

This PR removes passing down the schema to let Arrow decide unless:

• The type should be evolved
• In case of re-ordering, we reorder the original types

Let me know what you think @syun64

[kevinjqliu]

LGTM

[kevinjqliu]

Docs for promote_options="permissive"
https://arrow.apache.org/docs/python/generated/pyarrow.concat_tables.html

[Fokko]

I'm aware of the failing CI. Looking into this. It looks like we can automatically cast in the RecordBatchReader.

[sungwy]

Hi @Fokko - thank you very much for taking another look into this issue. I've left some comments on what I've found to be issues in following this approach when I was working on the original PR.

[sungwy]

When I had originally worked on #786 I thought of this approach as well, but ran into issues like:

tests/integration/test_reads.py::test_pyarrow_batches_deletes[session_catalog_hive] - pyarrow.lib.ArrowTypeError: Field 0 cannot be cast from date32[day] to date32[day]

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "pyarrow/ipc.pxi", line 800, in pyarrow.lib.RecordBatchReader.cast
  File "pyarrow/error.pxi", line 154, in pyarrow.lib.pyarrow_internal_check_status
  File "pyarrow/error.pxi", line 91, in pyarrow.lib.check_status
pyarrow.lib.ArrowTypeError: Field 0 cannot be cast from date32[day] to date32[day]

As a workaround, I opted to cast each pa.Array individually within to_requested_schema, rather than using this API.

This issue is fixed in apache/arrow#41884, but until we use the new release, I don't think we will be able to use this approach

[Fokko]

Nice, good to see that it has been fixed. I was also pushing a patch: apache/arrow#43183 The RC0 of Arrow 17 has been cut, so the release should be there anytime soon

[sungwy]

Here, we are making an opinionated decision on whether we are using large or small type as the pyarrow schema when reading the Iceberg table as a RecordBatchReader. Is there a reason why we don't want to do the same for the table API? I've noticed that we've changed the return type of the Table API to Optional[pa.Table] in order to avoid having to use schema_to_pyarrow.

Similarly, other libraries like polars use the approach of choosing one type over the other (large types in the case of polars).

>>> strings = pa.array(["a", "b"])
>>> pydict = {"strings": strings}
>>> pa.Table.from_pydict(pydict)
pyarrow.Table
strings: string
----
strings: [["a","b"]]
>>> pq.write_table(pa.Table.from_pydict(pydict), "strings.parquet")
>>> pldf = pl.read_parquet("strings.parquet", use_pyarrow=True)
>>> pldf.dtypes
[String]
>>> pldf.to_arrow()
pyarrow.Table
strings: large_string
----
strings: [["a","b"]]

[Fokko]

My preference would be to let Arrow decide. For Polars it is different because they are also the query engine. Casting the types will recompute the buffers, consuming additional memory/CPU, which I would rather avoid.

For the table, we first materialize all the batches in memory, so if one of them is large, it will automatically upcast, otherwise, it will keep the small types.

[sungwy]

My knowledge on Parquet data to Arrow buffer conversion is less versed, so please do check me if I am not making much sense 🙂

But are we actually casting the types on read?

We make a decision on whether we are choosing to read with large or small types when instantiating the fragment scanner, which loads the parquet data into the Arrow buffers. The schema_to_pyarrow() calls to pa.Table or pa.RecordBatchReader or in to_requested_schema following that all represent the Table schema in the consistent (large or small) format which shouldn't result in any additional casting and reassigning of buffers.

I think the only time we are casting the types is on write, where we may want to downcast it for forward compatibility. It looks like we have to choose a schema to use on write anyways, because using a schema for the ParquetWriter that isn't consistent with the schema within the dataframe results in an exception.

[HonahX]

I think the only time we are casting the types is on write, where we may want to downcast it for forward compatibility.

+1 Currently, we use "large_*" types during write. I think it could be better if we can write file based on the input pyarrow dataframe schema: if the dataframe is string, we also write with string

[sungwy]

Ah good catch on this one as well 👍

[sungwy]

🥇 LGTM - thank you @Fokko for putting this together

[HonahX]

LGTM!

[HonahX]

I think the only time we are casting the types is on write, where we may want to downcast it for forward compatibility.

+1 Currently, we use "large_*" types during write. I think it could be better if we can write file based on the input pyarrow dataframe schema: if the dataframe is string, we also write with string