Integrate DataFusion as execution engine for compute-heavy operations

[qzyu999]

Feature Request / Improvement

Problem

PyIceberg cannot perform several operations at production scale due to unbounded memory requirements in the PyArrow execution path:

• Tables with equality deletes are unreadable (hard ValueError)
• CoW deletes OOM on large Parquet files (~1GB)
• CoW overwrite OOMs (same pattern as delete)
• Upsert uses O(n²) row-by-row comparison
• Compaction not implemented, requires external sort (infeasible in-memory for large tables)
• Orphan file deletion OOMs (LEFT ANTI JOIN of millions of paths)

The list goes on (documented below) with operations that don't scale in the typical single-node environment of PyIceberg. These block PyIceberg from achieving feature parity with Java Iceberg for V2/V3.

Proposed Solution

Integrate Apache DataFusion as an optional execution engine (pip install 'pyiceberg[pyiceberg-core]') behind an automatic engine-resolution layer. When installed, compute-heavy operations use DataFusion's spill-to-disk execution (bounded memory). When not installed, the existing PyArrow path remains unchanged (works for small data, OOMs gracefully on large).

No existing behavior changes. No forced dependency. DuckDB-style UX where only a developer only needs to configure a memory budget if they so choose.

Design Doc

Support for PyIceberg DataFusion Integration

Operations Unblocked

• Equality delete read resolution
• Streaming CoW delete/overwrite
• Table compaction (sort + rewrite)
• Orphan file deletion
• Upsert via hash join
• Equality-to-positional conversion
• Position delete compaction
• Full MoR compaction
• Z-Order / Hilbert sorting
• DV compaction
• Incremental compaction
• Sort-order enforcement on write
• Dynamic partition overwrite (bounded memory)

Related Issues

PyIceberg

• [Feat] Support Merge-on-Read mode for Deletes #1078 (MoR support epic)
• [feature request] Support reading equality delete files #1210 / Equality Delete support #3270 (equality delete reads)
• Remove pyiceberg-core from the pyarrow extra #3356 (execution path isolation)
• Support data files compaction #1092 (data compaction)
• Delete orphan files #1200 (orphan file deletion)
• Add support for Equality Deletes on DeleteFileIndex #3285 (DeleteFileIndex for equality deletes)
• Add commit retry with data conflict validation #3319 / Add commit retry and concurrency validation for writes #3320 (commit retry, prerequisite for safe compaction commits)
• Feature: Add metadata-only replace API to Table for REPLACE snapshot operations #3130 / feat: Add metadata-only replace API to Table for REPLACE snapshot operations #3131 (REPLACE API, prerequisite for compaction)
• V3 Tracking issue #1818 (V3 tracking, DV compaction)
• Support producing positional deletes #1808 (positional delete write support)
• feat: Add DeleteFileIndex to improve position delete lookup #2918 (DeleteFileIndex for positional deletes, merged foundation)

iceberg-rust

• iceberg-rust#2186 (MoR scan-side delete reconciliation)
• iceberg-rust#2205 (equality delete reader)
• iceberg-rust#1530 (delete file support in scan)
• iceberg-rust#2269 (DataFusion write actions)

datafusion-python

• datafusion-python#1217 (FFI boundary stability)

[qzyu999]

I'm planning to start on the first sub-task (engine resolution module).