`Table.add_files` fails for Parquet files with `DecimalType` columns stored as `FIXED_LEN_BYTE_ARRAY` when precision allows `INT32`/`INT64`

[CaptainEureka]

Apache Iceberg version

0.9.1 (latest release)

Please describe the bug 🐞

When attempting to add Parquet files to an Iceberg table using Table.add_files, the operation fails if a column defined as DecimalType in the Iceberg schema is physically stored as FIXED_LEN_BYTE_ARRAY in the Parquet file, even if the decimal's precision would typically map to INT32 or INT64 according to Iceberg's preferred Parquet mapping.

I see in the Iceberg Spec that on-write the mapping is correct. However, the current behaviour seems to overly restrict the physical Parquet type for decimals during the file addition process. I believe this greatly limits the kinds of parquet files that can be "added" to an Iceberg table this way.

Steps to Reproduce:

1. Define an Iceberg table schema with a DecimalType column, for example, Decimal(10, 2).
   • Iceberg's preferred Parquet physical type for Decimal(10, 2) would be INT64.
2. Create a Parquet file where the corresponding column for this Decimal(10, 2) is physically stored as FIXED_LEN_BYTE_ARRAY. The data itself is valid for Decimal(10, 2).
3. Attempt to add this Parquet file to the Iceberg table using Table.add_files.

Behavior:

The Table.add_files operation fails, with the following error:

ValueError: Unexpected physical type FIXED_LEN_BYTE_ARRAY for DecimalType(10, 2) expected INT32

indicating a mismatch between the expected physical type (e.g., INT64) and the actual physical type (FIXED_LEN_BYTE_ARRAY) found in the Parquet file for the decimal column.

Expected Behavior:

The Table.add_files operation should succeed and correctly read the decimal values from the FIXED_LEN_BYTE_ARRAY physical storage. The Iceberg reader/writer should be lenient with the physical storage format of decimals OR otherwise Table.add_files should document these limitations.

Environment:

• Python version: 3.12.9
• Parquet library and version: pyarrow 20.0.0

P.S. If this is just user error and I shouldn't be trying to do things this way I'd be happy to hear alternatives.

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

[basil-soma]

I'm running into the same issue myself with table upserts since 0.9.1. Thanks for raising this!

[kevinjqliu]

I think this was fixed recently by #1839 which is not yet released
We have a nightly build running in https://test.pypi.org/project/pyiceberg/ which should include this change. Could you give that a try?

[basil-soma]

As of at least 0.10.0.dev20250605002910 this is resolved for me with the upsert. I have not replicated this with the original issue though. Thank you!

[CaptainEureka]

As of at least 0.10.0.dev20250605002910 this is resolved for me with the upsert. I have not replicated this with the original issue though. Thank you!

I tested this with build 0.10.0.dev20250606002903 and from main and it fails still on Table.add_files

[jayceslesar]

Can you provide a full script that can reproduce the issue?

[thijsheijden]

I am experiencing this bug as well. I am trying to insert the TPC-H dataset generated using DuckDB, written to Parquet files, into Iceberg using the add_files method. I have attached my Parquet generation code as well as my Iceberg creation code. I am using the abovementioned nightly build pyiceberg-0.10.0.dev20250630003157.

Generation of TPC-H Parquet files:

# Connect and load TPCH extension
con = duckdb.connect()
con.execute("INSTALL tpch; LOAD tpch")

# Define TPCH table names
tpch_tables = ['orders', 'lineitem']

# Optional: Set scale factor (controls number of rows)
scale = 1.0
con.execute(f"CALL dbgen(sf={scale})")

for table in tpch_tables:
    print(f'Generating files for table {table}')
    cur_offset = 0
    file_idx = 0
    batch_idx = 0
    while True:
        if file_idx % args.b == 0:
            Path(f'{table}/batch_{batch_idx}').mkdir(exist_ok=True)
            batch_idx += 1
            print(f'Writing batch {batch_idx}')

        query = f"SELECT * FROM '{table}' LIMIT 1000 OFFSET {cur_offset}"
        df = con.execute(query).arrow()  # Get as Arrow Table

        # Write to a single Parquet file
        pq.write_table(df, f'{table}/batch_{batch_idx - 1}/part{file_idx}.parquet')

        if len(df) < 1000:
            break

        file_idx += 1
        cur_offset += 1000

Insertion into Iceberg table:

    warehouse_path = "/data"
    catalog = load_catalog(
        "pyiceberg",
        **{
            'type': 'sql',
            "uri": f"sqlite:///{warehouse_path}/pyiceberg_catalog.db",
            "warehouse": f"file://{warehouse_path}",
        },
    )
    catalog.create_namespace_if_not_exists("default")

    # Load the batches of files to import
    batches = os.listdir(args.file_dir)
    first_file = os.path.join(args.file_dir, "batch_0", os.listdir(os.path.join(args.file_dir, "batch_0"))[0])

    # Create table using schema of the first file
    df = pq.read_table(first_file)
    table = catalog.create_table_if_not_exists(
        f"default.{args.table}",
        schema=df.schema,
    )

    batch_idx = 1
    for batch_dir in batches:
        print(f"Adding batch {batch_idx}")
        batch_dir =  os.path.join(args.file_dir, batch_dir)
        file_paths = os.listdir(batch_dir)
        file_paths = [batch_dir + '/' + s for s in file_paths]
        table.add_files(file_paths=file_paths, check_duplicate_files=False)
        batch_idx += 1

[CaptainEureka]

@thijsheijden This is essentially what I was trying to do as well. Thanks for the example script!