[I] Decompose io/pyarrow.py into focused modules to enable pluggable compute engines [iceberg-python]

["qzyu999 (via GitHub)" <gi...@apache.org>]

qzyu999 opened a new issue, #3737:
URL: https://github.com/apache/iceberg-python/issues/3737

   ## Summary
   
   `pyiceberg/io/pyarrow.py` is a 3,100+ line monolith that handles six unrelated concerns: filesystem I/O, schema conversion, expression translation, scan/read orchestration, write logic, and Parquet statistics. This makes it difficult to test individual components, extend behavior, or substitute alternative engines for specific operations.
   
   This issue proposes an incremental decomposition - a series of small, independently-reviewable refactoring PRs that split the file by concern while maintaining full backward compatibility via re-exports. The end goal is clean seam points where bounded-memory compute engines (DataFusion, etc.) can be introduced for operations that currently OOM on large data.
   
   ## Motivation
   
   Several open issues depend on bounded-memory compute that PyArrow's kernel library cannot provide:
   
   - Equality delete resolution (#1210, #3270) - requires anti-join with spill
   - Sort-on-write - requires external merge sort
   - Data compaction (#1092) - requires sort + join + rewrite pipeline
   - CoW deletes on large files - full file materialization causes OOM
   
   An earlier attempt to address this (pluggable backend PR) was rejected as too large. This issue takes the opposite approach: decompose first, add capabilities later.
   
   ## Approach
   
   ### Phase 1: Split the monolith (pure refactoring)
   
   Extract each concern into its own module under `pyiceberg/io/`. The original `pyarrow.py` becomes a thin re-export shim so all existing imports continue to work.
   
   | PR | Extraction | Approximate scope |
   |----|-----------|-------------------|
   | A | FileIO (`PyArrowFile`, `PyArrowFileIO`) | ~700 lines |
   | B | Schema conversion (`schema_to_pyarrow`, `pyarrow_to_schema`, visitors) | ~900 lines |
   | C | Expression translation (`expression_to_pyarrow`, `_ConvertToArrowExpression`) | ~300 lines |
   | D | Statistics (`StatsAggregator`, `PyArrowStatisticsCollector`, `ParquetFormatWriter`) | ~500 lines |
   | E | Write path (`write_file`, `_dataframe_to_data_files`, partitioning, bin packing) | ~1200 lines |
   | F | Scan/Read (`ArrowScan`, `_task_to_record_batches`, delete resolution) | ~300 lines |
   
   Each PR:
   - Moves code, does not change behavior
   - Re-exports from the original module path
   - All existing tests pass unchanged
   - No new dependencies
   
   These PRs are largely independent of each other (no strict ordering required).
   
   ### Phase 2: Introduce a compute protocol
   
   Once concerns are separated, introduce a thin `ComputeEngine` protocol for the operations that benefit from bounded-memory execution:
   
   `python
   class ComputeEngine(Protocol):
       def filter_batches(self, batches, expr, schema) -> Iterator[RecordBatch]: ...
       def sort_batches(self, batches, sort_order, schema) -> Iterator[RecordBatch]: ...
       def anti_join(self, left, right, keys) -> Iterator[RecordBatch]: ...
   `
   
   The default implementation delegates to the existing PyArrow code. No behavior change, just an indirection point.
   
   ### Phase 3: DataFusion as optional compute engine
   
   With the protocol in place, a `DataFusionComputeEngine` implementation slots in as an optional extra. Each capability (equality delete resolution, sort-on-write, etc.) is its own PR wiring the protocol into the specific code path.
   
   ## What this is NOT
   
   - Not a rewrite. Phase 1 is purely moving existing code into new files.
   - Not adding DataFusion as a hard dependency. It remains an optional extra.
   - Not changing the public API. All existing imports and behaviors are preserved.
   
   ## Prior art / references
   
   - Community sync discussion (June 30, 2026): established that read/write/compute should be separable
   - #3554: Original DataFusion integration proposal
   - #1210, #3270: Equality delete resolution
   - #1092: Data compaction
   
   ---
   
   I plan to start with PR A (FileIO extraction) as a proof of concept for the approach. Feedback on the overall direction is welcome before I proceed further.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [I] Decompose io/pyarrow.py into focused modules to enable pluggable compute engines [iceberg-python]

["rambleraptor (via GitHub)" <gi...@apache.org>]

rambleraptor commented on issue #3737:
URL: https://github.com/apache/iceberg-python/issues/3737#issuecomment-5332149146

   I think the idea of having pluggable implementations is really intriguing! The first thing that would be most useful is to ensure that there's no PyArrow logic that's occurring outside of pyarrow.py. After that, I'd love to help work through what the compute protocols would be. Ideally, we would help build those in tangent with building out the modules. 


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

Re: [I] Decompose io/pyarrow.py into focused modules to enable pluggable compute engines [iceberg-python]

["qzyu999 (via GitHub)" <gi...@apache.org>]

qzyu999 commented on issue #3737:
URL: https://github.com/apache/iceberg-python/issues/3737#issuecomment-5347164236

   > I think the idea of having pluggable implementations is really intriguing! The first thing that would be most useful is to ensure that there's no PyArrow logic that's occurring outside of pyarrow.py. After that, I'd love to help work through what the compute protocols would be. Ideally, we would help build those in tangent with building out the modules.
   
   Hi @rambleraptor, thanks for the feedback. I agree that consolidating PyArrow logic should come before decomposing the monolith.
   
   I did an audit of the `pyiceberg/` source (excluding `io/pyarrow.py` itself and tests) by grepping for:
   - Top-level `import pyarrow` / `from pyarrow import` statements
   - Inline runtime `import pyarrow as pa` inside function bodies
   - `TYPE_CHECKING`-only imports (these don't count as leakage since they're erased at runtime)
   
   Here's what I found:
   
   ### Runtime PyArrow usage outside `io/pyarrow.py`
   
   | Location | What it does | Notes |
   |----------|-------------|-------|
   | `table/upsert_util.py` | Top-level `import pyarrow as pa` + `from pyarrow import compute as pc`. Does joins, group_by, take, cast, slice directly on `pa.Table` | Clearest case for absorption. Pure compute logic that should route through the pyarrow module |
   | `table/inspect.py` | Runtime `import pyarrow as pa` in every method. Builds `pa.schema()`/`pa.Table.from_pylist()` for metadata tables (snapshots, entries, refs, partitions, manifests, etc.) | Gray area. This is output formatting, not compute that would benefit from engine substitution |
   | `transforms.py` | Runtime `import pyarrow as pa` in `_pyiceberg_transform_wrapper` and `VoidTransform.pyarrow_transform()`. Dispatches on `pa.Array` vs `pa.ChunkedArray` | Already explicitly scoped as pyarrow extension points on each transform class |
   | `table/__init__.py` | Runtime imports in `append()`, `overwrite()`, `dynamic_partition_overwrite()`, `upsert()`. Checks `isinstance(df, pa.Table)` then delegates to `io.pyarrow` | Intentional API surface. The heavy lifting already goes through `io.pyarrow` |
   | `table/deletion_vector.py` | Single `pa.chunked_array()` call in `_bitmaps_to_chunked_array` | Trivial |
   | `catalog/__init__.py` | `_convert_schema_if_needed()` with runtime import, but it delegates to `visit_pyarrow`/`_ConvertToIcebergWithoutIDs` from `io.pyarrow` | Already correctly structured |
   
   ### TYPE_CHECKING only (not leakage)
   
   `schema.py`, `table/puffin.py`, `table/update/schema.py`, `catalog/sql.py`, `catalog/rest/__init__.py`, `catalog/noop.py`, `catalog/hive.py`, `catalog/glue.py`, `catalog/dynamodb.py`, `catalog/bigquery_metastore.py` all import `pyarrow` under `TYPE_CHECKING` for type annotations only. No runtime dependency.
   
   ### Proposed plan
   
   I'd like to open a tracking issue for this and submit one PR per location, absorbing the PyArrow logic into `io/pyarrow.py` so that callers import helpers from `pyiceberg.io.pyarrow` rather than calling `pa.*` directly. Each PR is a pure refactor with no behavior change.
   
   My read on what to absorb vs. leave:
   
   - **Absorb:** `table/upsert_util.py` (joins, dedup, row comparison are all compute operations that should live behind the pyarrow module boundary)
   - **Discuss:** `table/inspect.py` (metadata table construction), `transforms.py` (pyarrow_transform dispatch)
   - **Leave:** `table/__init__.py` (user-facing API), `catalog/__init__.py` (already delegates correctly), `deletion_vector.py` (trivial)
   
   The goal is that once containment is done, every runtime PyArrow call routes through `io/pyarrow.py`. Then the decomposition (#3738) splits that one module into focused submodules, and the protocol/substitution work has clean seam points.
   
   Does this match what you had in mind? Happy to hear if any of the "leave" or "discuss" items should be treated differently.
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org