Error when filtering by UUID in table scan

[sevakva]

Apache Iceberg version

main (development)

Please describe the bug 🐞

Problem
Getting a pyarrow.lib.ArrowNotImplementedError: Function 'equal' has no kernel matching input types (extension<arrow.uuid>, extension<arrow.uuid>) when trying to scan a PyIceberg table with a row filter using UUID comparison. The error indicates that PyArrow's equal function doesn't have a kernel for comparing UUID extension types.

Environment
pyiceberg: Nightly build (expected to support UUIDs)
pyarrow: 21.0.0
Python: 3.13

Code to Reproduce

import uuid
from pyiceberg.expressions import EqualTo

# This fails with ArrowNotImplementedError
df = table.scan(row_filter=EqualTo("batch_id", uuid.UUID("0190de80-647f-4bbc-a80e-efda686b910f")))

Full Error Stack Trace

  File "/opt/homebrew/Cellar/python@3.13/3.13.3/Frameworks/Python.framework/Versions/3.13/lib/python3.13/concurrent/futures/_base.py", line 401, in __get_result
    raise self._exception
  File "/opt/homebrew/Cellar/python@3.13/3.13.3/Frameworks/Python.framework/Versions/3.13/lib/python3.13/concurrent/futures/thread.py", line 59, in run
    result = self.fn(*self.args, **self.kwargs)
  File "/Users/.venv/lib/python3.13/site-packages/pyiceberg/io/pyarrow.py", line 1694, in batches_for_task
    return list(self._record_batches_from_scan_tasks_and_deletes([task], deletes_per_file))
  File "/Users/.venv/lib/python3.13/site-packages/pyiceberg/io/pyarrow.py", line 1732, in _record_batches_from_scan_tasks_and_deletes
    for batch in batches:
                 ^^^^^^^
  File "/Users/.venv/lib/python3.13/site-packages/pyiceberg/io/pyarrow.py", line 1518, in _task_to_record_batches
    fragment_scanner = ds.Scanner.from_fragment(
        fragment=fragment,
    ...<4 lines>...
        columns=[col.name for col in file_project_schema.columns],
    )
  File "pyarrow/_dataset.pyx", line 3692, in pyarrow._dataset.Scanner.from_fragment
  File "pyarrow/_dataset.pyx", line 3458, in pyarrow._dataset._populate_builder
  File "pyarrow/_compute.pyx", line 2732, in pyarrow._compute._bind
  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.ArrowNotImplementedError: Function 'equal' has no kernel matching input types (extension<arrow.uuid>, extension<arrow.uuid>)

Expected Behavior
The table scan should successfully filter rows by UUID without throwing a kernel matching error.

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]

looks like we changed pyiceberg's internal uuid representation in https://github.com/apache/iceberg-python/pull/2007/files#diff-8d5e63f2a87ead8cebe2fd8ac5dcf2198d229f01e16bb9e06e21f7277c328abdR687 from pa.binary(16) to pa.uuid()

this is in the call path for scanning.
pyarrow_filter -> expression_to_pyarrow -> ConvertToArrowExpression.visit_equal -> _convert_scalar -> schema_to_pyarrow -> _ConvertToArrowSchema.visit_uuid

there are other unsupported functions in pyarrow for uuid. for example, apache/arrow#47094
and it is called out in #2007 that uuid extension type is not fully supported yet.

maybe we can revert back to using pa.binary(16) as the internal representation.

@Fokko wdyt?

[kevinjqliu]

i dug into this a bit more. the issue is not with pyiceberg's internal uuid representation, its with pushing down the pyarrow filter to pyarrow's ds.Scanner

iceberg-python/pyiceberg/io/pyarrow.py

Lines 1518 to 1525 in 19efd2d

	fragment_scanner = ds.Scanner.from_fragment(
	fragment=fragment,
	schema=physical_schema,
	# This will push down the query to Arrow.
	# But in case there are positional deletes, we have to apply them first
	filter=pyarrow_filter if not positional_deletes else None,
	columns=[col.name for col in file_project_schema.columns],
	)

breakpointing here shows

(Pdb) pyarrow_filter
<pyarrow.compute.Expression (uuid == "00000000000000000000000000000000")>

Heres the script to repro

from pyiceberg.catalog import load_catalog
import uuid
import pyarrow as pa
from pyiceberg.expressions import EqualTo

catalog = load_catalog("default", type="in-memory")
catalog.create_namespace_if_not_exists("default")

TEST_DATA_WITH_NULL = {
    "uuid": [
        uuid.UUID("00000000-0000-0000-0000-000000000000").bytes,
        None,
        uuid.UUID("11111111-1111-1111-1111-111111111111").bytes,
        uuid.UUID("22222222-2222-2222-2222-222222222222").bytes,
    ]
}

# schema = pa.schema([("uuid", pa.binary(16))])
schema = pa.schema([("uuid", pa.uuid())])

# Create table
arrow_table = pa.table(TEST_DATA_WITH_NULL, schema=schema)

table = catalog.create_table("default.test", schema=schema)
table.append(arrow_table)
# print(table.schema())

test_cases = [
    uuid.UUID("00000000-0000-0000-0000-000000000000").bytes,
    uuid.UUID("00000000-0000-0000-0000-000000000000"),
    # "00000000-0000-0000-0000-000000000000",
]

for test_case in test_cases:
   result = table.scan(row_filter=EqualTo("uuid", test_case)).to_arrow()
   assert len(result) == 1

[Fokko]

If we need to convert from uuid to fixed[16] to fix this, I think that would be reasonable.

[ndrluis]

@Fokko @kevinjqliu I started working on a fix, and while preparing to write a test, I noticed we already have a test that should cover this scenario. Through my testing, I found that the issue isn't with reading but with writing. Parquet files written by PySpark (parquet-mr version 1.15.2) work fine, but those written by PyIceberg don't. I tried updating PyArrow to the latest version thinking it might be a missing logical type issue, but that didn't fix it. There seems to be something different about how parquet-mr handles UUID writing.

Given this, I think we should revert to fixed[16], though we'll likely run into the same problem @Fokko fixed in #2007. If I'm understanding correctly, that was a logical type issue resolved in PyArrow 21.0.0. So reverting to fixed[16]and updating to PyArrow 21.0.0 should solve this for now.

The catch is that Bodo has a hard requirement for PyArrow 19.0. I think this deserves its own discussion, but I'd love to hear your thoughts. Maybe we should start a thread on the mailing list about these kinds of dependency constraints. While I get that having to_bodo() just work out of the box is nice for users, I feel like Bodo should be the one supporting Iceberg, not PyIceberg bending over backwards to support Bodo.

[thoec]

Any updates on this?

[kevinjqliu]

The catch is that Bodo has a hard requirement for PyArrow 19.0. I think this deserves its own discussion, but I'd love to hear your thoughts. Maybe we should start a thread on the mailing list about these kinds of dependency constraints. While I get that having to_bodo() just work out of the box is nice for users, I feel like Bodo should be the one supporting Iceberg, not PyIceberg bending over backwards to support Bodo.

yes i agree. Do you mind starting a thread on this? For additional context, I started a PR in bodo to remove the upperbound constraint bodo-ai/Bodo#794

[ndrluis]

Any updates on this?

@thoec I was on vacation and I will continue debugging this to understand the differences between the Java and C++ Arrow-written Parquet files. From the Parquet metadata level, if we update PyArrow to version 21, we get the same result, but the scan error persists. I have thought of some strategies to debug this, but I will only have time to dive deeper into this tomorrow.

yes i agree. Do you mind starting a thread on this? For additional context, I started a PR in bodo to remove the upperbound constraint bodo-ai/Bodo#794

@kevinjqliu I will start a thread on the mailing list

[ndrluis]

@Fokko @kevinjqliu I made the changes, but during my tests I received an error from the Java implementation that I'm fixing with this PR apache/iceberg#14027. So to maintain compatibility between PyIceberg and the Java implementation, we need this fix.

The test that returns the error (before changing the visit_uuid to binary(16)):

def test_write_uuid_in_pyiceberg_and_scan(session_catalog: Catalog, spark: SparkSession) -> None:
    identifier = "default.test_write_uuid_in_pyiceberg_and_scan"

    catalog = load_catalog("default", type="in-memory")
    catalog.create_namespace("ns")

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

    test_data_with_null = {
        "uuid_col": [
            uuid.UUID("00000000-0000-0000-0000-000000000000").bytes,
            None,
            uuid.UUID("11111111-1111-1111-1111-111111111111").bytes,
        ]
    }

    try:
        session_catalog.drop_table(identifier=identifier)
    except NoSuchTableError:
        pass

    table = _create_table(session_catalog, identifier, {"format-version": "2"}, schema=schema)

    arrow_table = pa.table(test_data_with_null, schema=schema.as_arrow())

    # Write with pyarrow
    table.append(arrow_table)

    # Write with pyspark
    spark.sql(
        f"""
        INSERT INTO {identifier} VALUES ("22222222-2222-2222-2222-222222222222")
        """
    )
    df = spark.table(identifier)

    table.refresh()

    assert df.count() == 4
    assert len(table.scan().to_arrow()) == 4

    result = df.where("uuid_col = '00000000-0000-0000-0000-000000000000'")
    assert result.count() == 1

    result = df.where("uuid_col = '22222222-2222-2222-2222-222222222222'")
    assert result.count() == 1

    result = table.scan(row_filter=EqualTo("uuid_col", uuid.UUID("00000000-0000-0000-0000-000000000000").bytes)).to_arrow()
    assert len(result) == 1

    result = table.scan(row_filter=EqualTo("uuid_col", uuid.UUID("22222222-2222-2222-2222-222222222222").bytes)).to_arrow()
    assert len(result) == 1