[Bug] Error in overwrite(): pyarrow.lib.ArrowInvalid: offset overflow with large dataset (~3M rows)

[sundaresanr]

Apache Iceberg version

0.8.1 (latest release)

Please describe the bug 🐞

Encountered the following error while calling overwrite() on a dataset with over 3 million rows (~1GB parquet file size; ~6 GB pyarrow table size):

pyarrow.lib.ArrowInvalid: offset overflow while concatenating arrays

Backtrace:

txn.overwrite(pat, overwrite_filter=overwrite_filter)
  File ".../site-packages/pyiceberg/table/__init__.py", line 470, in overwrite
    for data_file in data_files:
  File ".../site-packages/pyiceberg/io/pyarrow.py", line 2636, in _dataframe_to_data_files
    partitions = _determine_partitions(spec=table_metadata.spec(), schema=table_metadata.schema(), arrow_table=df)
                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File ".../site-packages/pyiceberg/io/pyarrow.py", line 2726, in _determine_partitions
    arrow_table = arrow_table.take(sort_indices)
                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "pyarrow/table.pxi", line 2133, in pyarrow.lib._Tabular.take
  File ".../site-packages/pyarrow/compute.py", line 487, in take
    return call_function('take', [data, indices], options, memory_pool)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "pyarrow/_compute.pyx", line 590, in pyarrow._compute.call_function
  File "pyarrow/_compute.pyx", line 385, in pyarrow._compute.Function.call
  File "pyarrow/error.pxi", line 155, in pyarrow.lib.pyarrow_internal_check_status
  File "pyarrow/error.pxi", line 92, in pyarrow.lib.check_status
pyarrow.lib.ArrowInvalid: offset overflow while concatenating arrays

Seems to be related to:

apache/arrow#25822
apache/arrow#33049

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

[kevinjqliu]

thanks for reporting this issue @sundaresanr do you have an example so i can reproduce this issue?

[kevinjqliu]

I've recreated this example from apache/arrow#33049 and was able to append successfully.

import numpy as np
import pyarrow as pa

from pyiceberg.catalog.sql import SqlCatalog

# Create a large string
x = str(np.random.randint(low=0, high=1000, size=(30000,)).tolist())
t = pa.chunked_array([x] * 20_000)
# Create pa.table
arrow_table = pa.table({"a": t})

warehouse_path = "/tmp/warehouse"
catalog = SqlCatalog(
    "default",
    uri=f"sqlite:///{warehouse_path}/pyiceberg_catalog.db", warehouse=f"file://{warehouse_path}",
)
catalog.create_namespace_if_not_exists("default")
try:
    catalog.drop_table("default.locations")
except:
    pass
table = catalog.create_table("default.locations", arrow_table.schema)
table.append(arrow_table)

[sundaresanr]

@kevinjqliu The issue can be reproduced with a partitioned table.

As you can see in the backtrace, the issue is with the call to _determine_partitions(), which calls arrow_table.take()

You can reproduce the issue with this code [added partition spec]

import numpy as np
import pyarrow as pa

from pyiceberg import transforms
from pyiceberg.catalog.sql import SqlCatalog

x = np.random.randint(low=0, high=1000, size=(30_000,)).tolist()
ta = pa.chunked_array([x] * 10_000)
y = ["fixed_string"] * 30_000
tb = pa.chunked_array([y] * 10_000)
# Create pa.table
arrow_table = pa.table({"a": ta, "b": tb})

warehouse_path = "/tmp/warehouse"
catalog = SqlCatalog(
    "default",
    uri=f"sqlite:///{warehouse_path}/pyiceberg_catalog.db", warehouse=f"file://{warehouse_path}",
)
catalog.create_namespace_if_not_exists("default")
try:
    catalog.drop_table("default.locations")
except:
    pass
table = catalog.create_table("default.locations", arrow_table.schema)
with table.update_spec() as update_spec:
    update_spec.add_field("b", transforms.IdentityTransform(), "pb")
table.append(arrow_table)

[bigluck]

We're facing the same problem by using pyiceberg 0.7.1 :(

[Fokko]

I can take a stab at this if nobody objects.

@bigluck Why are you still at 0.7.1? 😆

[bigluck]

@Fokko I'm getting older :D
BTW I think there's an easy fix. This comment is pretty interesting:

apache/arrow#33049 (comment)

Yes, it essentially is. We use 32-bit offsets for the default string and binary types. This means you always have to be careful when generating the arrays to chunk them if they get too large. Or otherwise use the LargeString variant, which uses 64-bit offsets. Polars, for example, takes the latter approach; they always use LargeString.

We tested it and it works. In our case we casted all the string types into large_string before passing the data to pyiceberg.

Knowing pyiceberg is already casting (during a scan operation) all the string into large_string, I think it's ok if pyiceberg does the same thing before writing the data into the lake (before passing the data to arrow).

It should be a noop (this code does not cover all the cases):

    def upcast_to_large_types(old_schema: pa.Schema) -> pa.Schema:
        fields = []
        for field in old_schema:
            if pt.is_string(field.type):
                fields.append(pa.field(field.name, pa.large_string()))
            elif pt.is_binary(field.type):
                fields.append(pa.field(field.name, pa.large_binary()))
            else:
                fields.append(field)

        return pa.schema(fields)

[Fokko]

@bigluck I was suspecting something like that, thanks for testing that out. One of the issues is that we combine the Array into a single one, which can easily go over the maximum offset. Let me see if there is a way without having bump everything to large_.

[Fokko]

@bigluck I got a solution that fixes the issue: #1539 LMKWYT