`to_arrow_batch_reader` returns a different schema than `to_arrow`

[enkidulan]

Apache Iceberg version

main (development)

Please describe the bug 🐞

In the development version, I noticed that the to_arrow_batch_reader method casts all string types to large_string, whereas the to_arrow method returns the schema as defined in the parquet file. At first glance, it looks like a bug, likely a regression from #1669

Here is a script you can use to reproduce the issue:

import pyarrow as pa
from pyiceberg.catalog import load_catalog
from uuid import uuid4
from pyiceberg.schema import Schema
from pyiceberg.types import NestedField, StringType, DoubleType

catalog = load_catalog("default")

df = pa.Table.from_pylist(
    [
        {"city": "Amsterdam", "lat": 52.371807, "long": 4.896029},
        {"city": "San Francisco", "lat": 37.773972, "long": -122.431297},
        {"city": "Drachten", "lat": 53.11254, "long": 6.0989},
        {"city": "Paris", "lat": 48.864716, "long": 2.349014},
    ],
)

schema = Schema(
    NestedField(1, "city", StringType(), required=False),
    NestedField(2, "lat", DoubleType(), required=False),
    NestedField(3, "long", DoubleType(), required=False),
)

tbl = catalog.create_table(f"default.cities-{uuid4()}", schema=schema)

tbl.overwrite(df)


schema_to_arrow = tbl.scan().to_arrow().schema

schema_to_arrow_batch_reader = tbl.scan().to_arrow_batch_reader().schema

print("schema_to_arrow == schema_to_arrow_batch_reader", schema_to_arrow == schema_to_arrow_batch_reader)
print("\nschema_to_arrow")
print(schema_to_arrow)
print("\nschema_to_arrow_batch_reader")
print(schema_to_arrow_batch_reader)

output:

schema_to_arrow == schema_to_arrow_batch_reader False

schema_to_arrow:
city: string
lat: double
long: double

schema_to_arrow_batch_reader:
city: large_string                               
  -- field metadata --
  PARQUET:field_id: '1'
lat: double
  -- field metadata --
  PARQUET:field_id: '2'
long: double
  -- field metadata --
  PARQUET:field_id: '3'

Notice that in to_arrow schema says city: string, while in to_arrow_batch_reader it's city: large_string

Willingness to contribute

• I can contribute a fix for this bug independently
• I would be willing to contribute a fix for this bug with guidance from the Iceberg community
• I cannot contribute a fix for this bug at this time

[Fokko]

Some context, as this is a bit of a loose-end:

• When we read using to_arrow() we pull all the data in memory right away, so we know that all the types of the column.
• When reading this through the to_arrow_batch_reader() we read the files one by one, and we might hit a Parquet file that uses large_string rather than string. We cannot push it into the string since it would cause a buffer overflow. Therefore we're taking the safe path and we upcast it into a large_string upfront.

Out of curiousity, what is your expected outcome here?

[enkidulan]

Thanks for getting back to me.

As for the expectation, I'd expect to_arrow_batch_reader to return the type defined in the table schema. Why would any of the parquet files in the table not be compatible with the table schema?

---

However, there is another aspect to this. After upgrading to the upstream version, table.scan().to_arrow() call started to fail on existing tables with the following traceback:

*** pyarrow.lib.ArrowTypeError: Unable to merge: Field XXXXX has incompatible types: string vs dictionary<values=string, indices=int32, ordered=0>
Traceback (most recent call last):
  File ".../pyiceberg/table/__init__.py", line 1982, in to_arrow
    ).to_table(self.plan_files())
      ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File ".../pyiceberg/io/pyarrow.py", line 1623, in to_table
    result = pa.concat_tables(
             ^^^^^^^^^^^^^^^^^
  File "pyarrow/table.pxi", line 6256, in pyarrow.lib.concat_tables
  File "pyarrow/error.pxi", line 155, in pyarrow.lib.pyarrow_internal_check_status
  File "pyarrow/error.pxi", line 92, in pyarrow.lib.check_status

We are doing recoding using to_arrow_batch_reader function and then writing results with table.append back to the table. From table.append point of view, the type that to_arrow_batch_reader returns is backward compatible; however, pyarrow fails on trying to read it, as it gets two parquet files with incompatible schemas (from its point of view). To be fair, we specify both pyiceberd and pyarrow schemas explicitly, but I didn't expect table.append to succeed if the table schema is not compatible with other parquet files. So this is the point of conduction for me: to_arrow_batch_reader and to_arrow were returning different types than specified in the table schema.

Here is the code you can reproduce this issue with:

from pyiceberg.catalog import load_catalog
from pyiceberg.schema import Schema
from pyiceberg.types import NestedField, StringType
import pyarrow as pa

# Create schema
pa_schema = pa.schema(
    [
        pa.field("name", pa.dictionary(pa.uint8(), pa.string())),
    ])

schema = Schema(
    NestedField(field_id=1, name="name", field_type=StringType(), required=False)
)

# Load catalog (using in-memory for simplicity)
catalog = load_catalog(
    "example",
    **{
        "uri": "http://127.0.0.1:8181",
        "s3.endpoint": "http://127.0.0.1:9000",
        "py-io-impl": "pyiceberg.io.pyarrow.PyArrowFileIO",
        "s3.access-key-id": "admin",
        "s3.secret-access-key": "password",
    }
)

# Create table
table = catalog.create_table(f"test.example", schema=schema)

# Create sample data in table
table.append(pa.table({"name": ["Alice", "Bob", "Charlie"]}, schema=pa_schema))

# this works fine
print("\n\nReading data from table")
print(table.scan().to_arrow())

# Read data using to_arrow_batch_reader and write to table
print("\n\nWriting data to table")
batch_reader = table.scan().to_arrow_batch_reader()
for batch in batch_reader:
    arrow_table = pa.Table.from_batches([batch])
    table.append(arrow_table)

# this fails
print("\n\nReading data from table")
print(table.scan().to_arrow())

[enkidulan]

I'm happy to provide more context on the issue or contribute to the solution, whatever it may be (if updating docs to manage users' expectations is the solution, I'm happy to create a PR)