qzyu999 opened a new pull request, #3716:
URL: https://github.com/apache/iceberg-python/pull/3716
## Which issue does this PR close?
Closes #3715
Closes #1210
Closes #3270
Closes #3554
## Rationale for this change
PyIceberg's I/O and compute logic lives in a single 3,000+ line file
(`pyiceberg/io/pyarrow.py`) with no separation of concerns. PyArrow is a kernel
library without memory management or spill-to-disk, causing OOM crashes on
production-scale operations (CoW deletes, equality delete resolution, scan
planning).
This PR introduces a pluggable interface that:
1. Decouples PyIceberg from PyArrow, enabling alternative backends
2. Integrates DataFusion for bounded-memory compute with spill-to-disk
3. Fixes OOM patterns while maintaining full backward compatibility
## What changes are included in this PR?
### New Modules (`pyiceberg/execution/`)
| Module | Purpose |
|--------|---------|
| `protocol.py` | `ReadBackend`, `WriteBackend`, `ComputeBackend` protocol
definitions |
| `engine.py` | Backend resolution, instantiation, config helpers |
| `_orchestrate.py` | Per-task scan execution with delete resolution |
| `planning.py` | `InMemoryPlanner` + `BoundedMemoryPlanner` |
| `backends/pyarrow_backend.py` | Default implementation (always available) |
| `backends/datafusion_backend.py` | Bounded-memory implementation
(optional) |
### Operations Delivered
| Operation | Before | After |
|-----------|--------|-------|
| Equality delete resolution | `ValueError` | ✅ Works (Grace Hash Join with
spill) |
| CoW delete (large files) | OOM | ✅ Two-pass streaming, O(batch_size)
memory |
| Positional deletes (millions) | OOM | ✅ Size-based routing to anti-join |
| Sort-on-write | Not implemented | ✅ External merge sort (best-effort) |
| Scan planning (>100K deletes) | OOM | ✅ BoundedMemoryPlanner with SQL join
|
### Migration
All data paths in `table/__init__.py` now route through `orchestrate_scan()`
instead of `ArrowScan`. The old `ArrowScan` class is deprecated (0.11.0) and
scheduled for removal in 0.12.0.
## Are there any user-facing changes?
**No breaking changes.** All existing APIs work identically.
**New behavior** (when `datafusion` is installed):
- Tables with equality deletes now return correct results (previously
`ValueError`)
- Large file operations complete without OOM
- Files are sorted on write if table defines a sort order
**New optional dependency**:
```bash
pip install 'pyiceberg[datafusion]'
```
When DataFusion is not installed, behavior is identical to before (may OOM
on large data).
**New configuration** (optional, sensible defaults):
```yaml
# .pyiceberg.yaml
execution:
compute-backend: datafusion # auto-detected when installed
memory-limit: 536870912 # 512 MB
planning-threshold: 100000 # switch to bounded planner
cow-threshold: 67108864 # 64 MB for two-pass streaming
```
## Testing
- `test_arrowscan_parity.py`: verifies new path matches deprecated ArrowScan
- `test_property_based.py`: Hypothesis tests for PyArrow/DataFusion
equivalence
- `test_positional_deletes.py`: DataFusion positional delete resolution
- `test_equality_deletes.py`: equality delete resolution with sequence
number gating
- `test_cow_streaming.py`: two-pass streaming for large files
- All existing integration tests pass without modification
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
