IO: Consolidate PyArrow logic into io/pyarrow.py before decomposition

[qzyu999]

Summary

Before decomposing pyiceberg/io/pyarrow.py into focused submodules (#3737, #3738), we should consolidate PyArrow-specific logic that currently lives outside the module. This ensures all PyArrow calls route through a single boundary, making the subsequent split clean and enabling future engine substitution.

Motivation

Per discussion in #3737, @rambleraptor noted that the first useful step is ensuring no PyArrow logic occurs outside pyarrow.py. Currently several modules import pyarrow directly and implement compute logic inline rather than delegating through pyiceberg.io.pyarrow.

When we later introduce a ComputeEngine protocol, any PyArrow logic outside the module boundary bypasses the protocol and prevents clean substitution.

Audit

Grepped pyiceberg/ (excluding io/pyarrow.py and tests) for runtime import pyarrow statements (both top-level and inline). Excluded TYPE_CHECKING-only imports since those have no runtime dependency.

Location	What it does	Action
table/upsert_util.py	PyArrow table joins, group_by, compute, cast, take	Absorb
table/inspect.py	Builds pa.schema + pa.Table.from_pylist for metadata inspection	TBD
transforms.py	pyarrow_transform() dispatch on pa.Array/ChunkedArray	TBD
table/__init__.py	Entry points accept pa.Table, delegate to io.pyarrow	Leave
table/deletion_vector.py	Single pa.chunked_array() call	Leave
catalog/__init__.py	Delegates to io.pyarrow for schema conversion	Leave

Plan

One PR per absorption. Each is a pure refactor: move code into io/pyarrow.py, have the caller import from pyiceberg.io.pyarrow instead of pyarrow directly. No behavior change, all existing tests pass unchanged.

• PR A: Absorb table/upsert_util.py PyArrow logic
• PR B: table/inspect.py (pending discussion)
• PR C: transforms.py (pending discussion)

Related

• Decompose io/pyarrow.py into focused modules to enable pluggable compute engines #3737 - Decompose io/pyarrow.py into focused modules
• IO: Extract PyArrowFileIO into dedicated module #3738 - Extract PyArrowFileIO (first decomposition step)
• Pluggable Backend Interface with DataFusion for Bounded-Memory Compute #3715 / Pluggable Backend Interface with DataFusion for Bounded-Memory Compute #3716 - Previous pluggable backend attempt (rejected as too large)

[qzyu999]

Notes on inspect.py and transforms.py

Adding some analysis on the two "TBD" items to guide discussion.

table/inspect.py

Every method in InspectTable follows the same pattern: iterate over metadata objects, build a list of Python dicts, define a pa.schema(...), and call pa.Table.from_pylist(data, schema=schema).

Absorption would mean moving the schema definitions and from_pylist calls into helper functions in io/pyarrow.py:

# io/pyarrow.py
def _build_snapshots_table(snapshots: list[dict[str, Any]]) -> pa.Table:
    schema = pa.schema([
        pa.field("committed_at", pa.timestamp(unit="ms"), nullable=False),
        pa.field("snapshot_id", pa.int64(), nullable=False),
        ...
    ])
    return pa.Table.from_pylist(snapshots, schema=schema)

Then inspect.py would only gather dicts and delegate:

# table/inspect.py
from pyiceberg.io.pyarrow import _build_snapshots_table

def snapshots(self) -> "pa.Table":
    rows = [{"committed_at": ..., ...} for s in self.tbl.metadata.snapshots]
    return _build_snapshots_table(rows)

My read: This is technically doable but adds ~10 trivial wrapper functions whose only job is pa.Table.from_pylist(data, schema=X). There's no compute here that benefits from engine substitution. A DataFusion engine would not build metadata inspection tables any differently. This is output formatting, not a substitution target

Leaning toward leave, but open to feedback.

transforms.py

Each transform subclass defines a pyarrow_transform() method returning a Callable[[pa.Array], pa.Array]. The shared _pyiceberg_transform_wrapper dispatches on pa.Array vs pa.ChunkedArray.

Absorption would mean moving the implementations out of the classes and into a dispatch function in io/pyarrow.py:

# io/pyarrow.py
def _get_pyarrow_transform(transform: Transform, source: IcebergType) -> Callable[[pa.Array], pa.Array]:
    if isinstance(transform, BucketTransform):
        return _pyiceberg_transform_wrapper(core.bucket, transform._num_buckets)
    elif isinstance(transform, YearTransform):
        return _pyiceberg_transform_wrapper(core.year, expected_type=pa.int32())
    ...

My read: This fights the existing OO design. The pyarrow_transform() methods are polymorphic for a reason: each transform subclass owns its own implementation. Pulling them into a central isinstance dispatch:

• Mirrors the class hierarchy in a fragile way (adding a new transform requires editing two places)
• Doesn't improve substitutability. If DataFusion came in, you'd add a parallel datafusion_transform() method or a generic transform(engine) dispatch alongside the existing methods, not replace them.

The methods are already scoped by name (pyarrow_transform) to signal they're the PyArrow-specific path. They are the extension point.

Leaning toward leave.

Summary

File	Absorption approach	Recommendation
inspect.py	Move schema defs + from_pylist into io/pyarrow.py helpers	Leave. No compute to substitute, just output formatting
transforms.py	Central dispatch function replacing polymorphic methods	Leave. Breaks OO design, methods already serve as the named extension point

Happy to reconsider if there's a use case I'm missing.